@Override
  protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir)
      throws Exception {
    Map<String, String> configs = pinotTaskConfig.getConfigs();
    String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY);
    String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
    String taskType = pinotTaskConfig.getTaskType();

    SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory();
    TableConfig tableConfig = getTableConfig(tableNameWithType);
    Schema schema = getSchema(tableNameWithType);
    SegmentPurger.RecordPurger recordPurger =
        recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(pinotTaskConfig, tableConfig, schema) : null;
    SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory();
    SegmentPurger.RecordModifier recordModifier =
        recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null;

    _eventObserver.notifyProgress(pinotTaskConfig, "Purging segment: " + indexDir);
    SegmentPurger segmentPurger =
        new SegmentPurger(indexDir, workingDir, tableConfig, schema, recordPurger, recordModifier);
    long purgeTaskStartTimeNs = MX_BEAN.getCurrentThreadCpuTime();
    File purgedSegmentFile = segmentPurger.purgeSegment();
    long purgeTaskEndTimeNs = MX_BEAN.getCurrentThreadCpuTime();
    _minionMetrics.addTimedTableValue(tableNameWithType, taskType, MinionTimer.TASK_THREAD_CPU_TIME_NS,
        purgeTaskEndTimeNs - purgeTaskStartTimeNs, TimeUnit.NANOSECONDS);
    if (purgedSegmentFile == null) {
      purgedSegmentFile = indexDir;
    }

    return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType)
        .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY))
        .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger())
        .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier())
        .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged())
        .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build();
  }
@Test
  public void testConvert()
      throws Exception {
    PurgeTaskExecutor purgeTaskExecutor = new PurgeTaskExecutor();
    purgeTaskExecutor.setMinionEventObserver(new MinionProgressObserver());
    PinotTaskConfig pinotTaskConfig = new PinotTaskConfig(MinionConstants.PurgeTask.TASK_TYPE, Collections
        .singletonMap(MinionConstants.TABLE_NAME_KEY, TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME)));
    File purgedIndexDir = purgeTaskExecutor.convert(pinotTaskConfig, _originalIndexDir, PURGED_SEGMENT_DIR).getFile();

    try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(purgedIndexDir)) {
      int numRecordsRemaining = 0;
      int numRecordsModified = 0;

      GenericRow row = new GenericRow();
      while (pinotSegmentRecordReader.hasNext()) {
        row = pinotSegmentRecordReader.next(row);
        numRecordsRemaining++;
        if (row.getValue(D1).equals(Integer.MAX_VALUE)) {
          numRecordsModified++;
        }
      }

      Assert.assertEquals(numRecordsRemaining, NUM_ROWS - 1);
      Assert.assertEquals(numRecordsModified, NUM_ROWS - 1);
    }
  }