public List<File> process()
      throws Exception {
    try {
      return doProcess();
    } catch (Exception e) {
      // Cleaning up output dir as processing has failed. file managers left from map or reduce phase will be cleaned
      // up in the respective phases.
      FileUtils.deleteQuietly(_segmentsOutputDir);
      throw e;
    } finally {
      FileUtils.deleteDirectory(_mapperOutputDir);
      FileUtils.deleteDirectory(_reducerOutputDir);
    }
  }
@Test
  public void testMultipleSegments()
      throws Exception {
    File workingDir = new File(TEMP_DIR, "multiple_segments_output");
    FileUtils.forceMkdir(workingDir);

    // Default configs
    SegmentProcessorConfig config =
        new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).build();
    SegmentProcessorFramework framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
    List<File> outputSegments = framework.process();
    assertEquals(outputSegments.size(), 1);
    String[] outputDirs = workingDir.list();
    assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
    SegmentMetadata segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
    assertEquals(segmentMetadata.getTotalDocs(), 10);
    assertEquals(segmentMetadata.getName(), "myTable_1597719600000_1597892400000_0");
    FileUtils.cleanDirectory(workingDir);
    rewindRecordReaders(_multipleSegments);

    // Time round, partition, rollup
    config = new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schema).setTimeHandlerConfig(
        new TimeHandlerConfig.Builder(TimeHandler.Type.EPOCH).setRoundBucketMs(86400000).setPartitionBucketMs(86400000)
            .build()).setMergeType(MergeType.ROLLUP).build();
    framework = new SegmentProcessorFramework(_multipleSegments, config, workingDir);
    outputSegments = framework.process();
    assertEquals(outputSegments.size(), 3);
    outputDirs = workingDir.list();
    assertTrue(outputDirs != null && outputDirs.length == 1, Arrays.toString(outputDirs));
    outputSegments.sort(null);
    segmentMetadata = new SegmentMetadataImpl(outputSegments.get(0));
    assertEquals(segmentMetadata.getTotalDocs(), 2);
    assertEquals(segmentMetadata.getName(), "myTable_1597708800000_1597708800000_0");
    segmentMetadata = new SegmentMetadataImpl(outputSegments.get(1));
    assertEquals(segmentMetadata.getTotalDocs(), 3);
    assertEquals(segmentMetadata.getName(), "myTable_1597795200000_1597795200000_1");
    segmentMetadata = new SegmentMetadataImpl(outputSegments.get(2));
    assertEquals(segmentMetadata.getTotalDocs(), 2);
    assertEquals(segmentMetadata.getName(), "myTable_1597881600000_1597881600000_2");
    FileUtils.cleanDirectory(workingDir);
    rewindRecordReaders(_multipleSegments);
  }