@Override
  public void run()
      throws Exception {
    // Get list of files to process.
    List<String> filteredFiles = SegmentGenerationUtils.listMatchedFilesWithRecursiveOption(_inputDirFS, _inputDirURI,
        _spec.getIncludeFileNamePattern(), _spec.getExcludeFileNamePattern(), _spec.isSearchRecursively());

    if (_consistentPushEnabled) {
      ConsistentDataPushUtils.configureSegmentPostfix(_spec);
    }

    File localTempDir = new File(FileUtils.getTempDirectory(), "pinot-" + UUID.randomUUID());
    try {
      int numInputFiles = filteredFiles.size();
      _segmentCreationTaskCountDownLatch = new CountDownLatch(numInputFiles);

      if (!SegmentGenerationJobUtils.useGlobalDirectorySequenceId(_spec.getSegmentNameGeneratorSpec())) {
        Map<String, List<String>> localDirIndex = new HashMap<>();
        for (String filteredFile : filteredFiles) {
          java.nio.file.Path filteredParentPath = Paths.get(filteredFile).getParent();
          localDirIndex.computeIfAbsent(filteredParentPath.toString(), k -> new ArrayList<>()).add(filteredFile);
        }
        for (String parentPath : localDirIndex.keySet()) {
          List<String> siblingFiles = localDirIndex.get(parentPath);
          Collections.sort(siblingFiles);
          for (int i = 0; i < siblingFiles.size(); i++) {
            URI inputFileURI = SegmentGenerationUtils
                .getFileURI(siblingFiles.get(i), SegmentGenerationUtils.getDirectoryURI(parentPath));
            submitSegmentGenTask(localTempDir, inputFileURI, i);
          }
        }
      } else {
        //iterate on the file list, for each
        for (int i = 0; i < numInputFiles; i++) {
          final URI inputFileURI = SegmentGenerationUtils.getFileURI(filteredFiles.get(i), _inputDirURI);
          submitSegmentGenTask(localTempDir, inputFileURI, i);
        }
      }
      _segmentCreationTaskCountDownLatch.await();

      if (_failure.get() != null) {
        _executorService.shutdownNow();
        throw _failure.get();
      }
    } finally {
      //clean up
      FileUtils.deleteQuietly(localTempDir);
      _executorService.shutdown();
    }
  }
@Test
  public void testSegmentGeneration() throws Exception {
    // TODO use common resource definitions & code shared with Hadoop unit test.
    // So probably need a pinot-batch-ingestion-common tests jar that we depend on.

    File testDir = makeTestDir();
    File inputDir = new File(testDir, "input");
    inputDir.mkdirs();
    File inputFile = new File(inputDir, "input.csv");
    FileUtils.writeLines(inputFile, Lists.newArrayList("col1,col2", "value1,1", "value2,2"));

    // Create an output directory, with two empty files in it. One we'll overwrite,
    // and one we'll leave alone.
    final String outputFilename = "myTable_OFFLINE_0.tar.gz";
    final String existingFilename = "myTable_OFFLINE_100.tar.gz";
    File outputDir = new File(testDir, "output");
    FileUtils.touch(new File(outputDir, outputFilename));
    FileUtils.touch(new File(outputDir, existingFilename));

    final String schemaName = "mySchema";
    File schemaFile = makeSchemaFile(testDir, schemaName);
    File tableConfigFile = makeTableConfigFile(testDir, schemaName);
    SegmentGenerationJobSpec jobSpec = makeJobSpec(inputDir, outputDir, schemaFile, tableConfigFile);
    jobSpec.setOverwriteOutput(false);
    SegmentGenerationJobRunner jobRunner = new SegmentGenerationJobRunner(jobSpec);
    jobRunner.run();

    // The output directory should still have the original file in it.
    File oldSegmentFile = new File(outputDir, existingFilename);
    Assert.assertTrue(oldSegmentFile.exists());

    // The output directory should have the original file in it (since we aren't overwriting)
    File newSegmentFile = new File(outputDir, outputFilename);
    Assert.assertTrue(newSegmentFile.exists());
    Assert.assertTrue(newSegmentFile.isFile());
    Assert.assertTrue(newSegmentFile.length() == 0);

    // Now run again, but this time with overwriting of output files, and confirm we got a valid segment file.
    jobSpec.setOverwriteOutput(true);
    jobRunner = new SegmentGenerationJobRunner(jobSpec);
    jobRunner.run();

    // The original file should still be there.
    Assert.assertTrue(oldSegmentFile.exists());

    Assert.assertTrue(newSegmentFile.exists());
    Assert.assertTrue(newSegmentFile.isFile());
    Assert.assertTrue(newSegmentFile.length() > 0);

    // FUTURE - validate contents of file?
  }