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 testRecordReaderFileConfigInit() throws Exception {
    File workingDir = new File(TEMP_DIR, "segmentOutput");
    FileUtils.forceMkdir(workingDir);
    ClassLoader classLoader = getClass().getClassLoader();
    URL resource = classLoader.getResource("data/dimBaseballTeams.csv");
    RecordReader recordReader = RecordReaderFactory.getRecordReader(FileFormat.CSV, new File(resource.toURI()),
        null, null);
    RecordReaderFileConfig recordReaderFileConfig = new RecordReaderFileConfig(FileFormat.CSV,
        new File(resource.toURI()),
        null, null, recordReader);
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName("myTable").
        setTimeColumnName("time").build();

    Schema schema =
        new Schema.SchemaBuilder().setSchemaName("mySchema").addSingleValueDimension("teamId",
                DataType.STRING, "")
            .addSingleValueDimension("teamName", DataType.STRING, "")
            .addDateTime("time", DataType.LONG, "1:MILLISECONDS:EPOCH", "1:MILLISECONDS").build();

    SegmentProcessorConfig config =
        new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema).build();
    SegmentProcessorFramework framework = new SegmentProcessorFramework(config, workingDir,
        ImmutableList.of(recordReaderFileConfig), Collections.emptyList(), null);
    List<File> outputSegments = framework.process();
    assertEquals(outputSegments.size(), 1);
    ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
    SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
    assertEquals(segmentMetadata.getTotalDocs(), 52);
    // Verify reader is closed
    assertEquals(recordReaderFileConfig.isRecordReaderClosedFromRecordReaderFileConfig(), true);
  }