@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 testFailureHandling()
      throws Exception {
    File testDir = makeTestDir();
    File inputDir = new File(testDir, "input");
    inputDir.mkdirs();

    File inputFile1 = new File(inputDir, "input1.csv");
    FileUtils.writeLines(inputFile1, Lists.newArrayList("col1,col2", "value11,11", "value12,12"));

    File inputFile2 = new File(inputDir, "input2.csv");
    FileUtils.writeLines(inputFile2, Lists.newArrayList("col1,col2", "value21,notanint", "value22,22"));

    File inputFile3 = new File(inputDir, "input3.csv");
    FileUtils.writeLines(inputFile3, Lists.newArrayList("col1,col2", "value31,31", "value32,32"));

    File outputDir = new File(testDir, "output");

    final String schemaName = "mySchema";
    File schemaFile = makeSchemaFile(testDir, schemaName);
    File tableConfigFile = makeTableConfigFile(testDir, schemaName);
    SegmentGenerationJobSpec jobSpec = makeJobSpec(inputDir, outputDir, schemaFile, tableConfigFile);

    // Set up for a segment name that matches our input filename, so we can validate
    // that only the first input file gets processed.
    SegmentNameGeneratorSpec nameSpec = new SegmentNameGeneratorSpec();
    nameSpec.setType(BatchConfigProperties.SegmentNameGeneratorType.INPUT_FILE);
    nameSpec.getConfigs().put(SegmentGenerationTaskRunner.FILE_PATH_PATTERN, ".+/(.+)\\.csv");
    nameSpec.getConfigs().put(SegmentGenerationTaskRunner.SEGMENT_NAME_TEMPLATE, "${filePathPattern:\\1}");
    jobSpec.setSegmentNameGeneratorSpec(nameSpec);

    try {
      SegmentGenerationJobRunner jobRunner = new SegmentGenerationJobRunner(jobSpec);
      jobRunner.run();
      fail("Job should have failed");
    } catch (Exception e) {
      assertTrue(e.getMessage().contains("input2.csv"), "Didn't find filename in exception message");

      // We should only have one output file, since segment generation will
      // terminate after the second input file.
      File[] segments = outputDir.listFiles(new FilenameFilter() {

        @Override
        public boolean accept(File dir, String name) {
          return name.endsWith(".tar.gz");
        }
      });

      // We rely on the SegmentGenerationJobRunner doing a sort by name, so "input1.csv" will be the
      // first file we process, and "input2.csv" (the bad file) will be the second one.
      assertEquals(segments.length, 1);
      assertTrue(segments[0].getName().endsWith("input1.tar.gz"));
    }
  }