public DataSchemaParser.ParseResult parseSources(String[] rawSources) throws IOException
  {
    Set<String> fileExtensions = _parserByFileExtension.keySet();
    Map<String, List<String>> byExtension = new HashMap<>(fileExtensions.size());
    for (String fileExtension : fileExtensions)
    {
      byExtension.put(fileExtension, new ArrayList<>());
    }

    String[] sortedSources = Arrays.copyOf(rawSources, rawSources.length);
    Arrays.sort(sortedSources);

    // Extract all schema files from the given source paths and group by extension (JARs are handled specially)
    for (String source : sortedSources)
    {
      final File sourceFile = new File(source);
      if (sourceFile.exists())
      {
        if (sourceFile.isDirectory())
        {
          // Source path is a directory, so recursively find all schema files contained therein
          final FileExtensionFilter filter = new FileExtensionFilter(fileExtensions);
          final List<File> sourceFilesInDirectory = FileUtil.listFiles(sourceFile, filter);
          // Add each schema to the corresponding extension's source list
          for (File f : sourceFilesInDirectory)
          {
            String ext = FilenameUtils.getExtension(f.getName());
            List<String> filesForExtension = byExtension.get(ext);
            if (filesForExtension != null)
            {
              filesForExtension.add(f.getAbsolutePath());
            }
          }
        }
        else if (sourceFile.getName().endsWith(".jar"))
        {
          // Source path is a JAR, so add it to each extension's source list.
          // The file-based parser for each extension will extract the JAR and process only files matching the extension
          byExtension.values().forEach(files -> files.add(sourceFile.getAbsolutePath()));
        }
        else
        {
          // Source path is a non-JAR file, so add it to the corresponding extension's source list
          String ext = FilenameUtils.getExtension(sourceFile.getName());
          List<String> filesForExtension = byExtension.get(ext);
          if (filesForExtension != null)
          {
            filesForExtension.add(sourceFile.getAbsolutePath());
          }
        }
      }
    }

    // Parse all schema files and JARs using the appropriate file format parser
    final ParseResult result = new ParseResult();
    for (Map.Entry<String, List<String>> entry : byExtension.entrySet())
    {
      String ext = entry.getKey();
      List<String> files = entry.getValue();
      _parserByFileExtension.get(ext).parseSources(files.toArray(new String[files.size()]), result);
    }

    return result;
  }
@Test(dataProvider = "inputFiles")
  public void testCustomSourceSchemaDirectory(String pegasusFilename, String[] expectedSchemas) throws Exception
  {
    String tempDirectoryPath = _tempDir.getAbsolutePath();
    String jarFile = tempDirectoryPath + FS + "test.jar";
    SchemaDirectory customSchemaDirectory = () -> "custom";
    String pegasusFile = TEST_RESOURCES_DIR + FS + pegasusFilename;
    String pegasusFileInJar = customSchemaDirectory.getName() + "/" + pegasusFilename;
    createTempJarFile(Collections.singletonMap(pegasusFile, pegasusFileInJar), jarFile);

    // Load with default parser, this will return zero scheams.
    DataSchemaParser parser = new DataSchemaParser.Builder(tempDirectoryPath).build();
    DataSchemaParser.ParseResult parseResult = parser.parseSources(new String[]{jarFile});
    assertEquals(parseResult.getSchemaAndLocations().size(), 0);

    // Now create a parser with custom directory as source
    parser = new DataSchemaParser.Builder(tempDirectoryPath)
        .setSourceDirectories(Collections.singletonList(customSchemaDirectory))
        .build();
    parseResult = parser.parseSources(new String[]{jarFile});
    assertEquals(parseResult.getSchemaAndLocations().size(), expectedSchemas.length);
    Set<String> schemaNames = parseResult.getSchemaAndLocations().keySet().stream().map(DataSchema::getUnionMemberKey).collect(
        Collectors.toSet());
    for (String schema : expectedSchemas)
    {
      assertTrue(schemaNames.contains(schema));
    }
    parseResult.getSchemaAndLocations().values().forEach(loc -> assertEquals(loc.getSourceFile().getAbsolutePath(), jarFile));
  }