public static long fix(FileSystem fs, Path dir,
                         Class<? extends Writable> keyClass,
                         Class<? extends Writable> valueClass, boolean dryrun,
                         Configuration conf) throws Exception {
    String dr = (dryrun ? "[DRY RUN ] " : "");
    Path data = new Path(dir, DATA_FILE_NAME);
    Path index = new Path(dir, INDEX_FILE_NAME);
    int indexInterval = conf.getInt(Writer.INDEX_INTERVAL, 128);
    if (!fs.exists(data)) {
      // there's nothing we can do to fix this!
      throw new Exception(dr + "Missing data file in " + dir + ", impossible to fix this.");
    }
    if (fs.exists(index)) {
      // no fixing needed
      return -1;
    }
    SequenceFile.Reader dataReader = 
      new SequenceFile.Reader(conf, SequenceFile.Reader.file(data));
    if (!dataReader.getKeyClass().equals(keyClass)) {
      throw new Exception(dr + "Wrong key class in " + dir + ", expected" + keyClass.getName() +
                          ", got " + dataReader.getKeyClass().getName());
    }
    if (!dataReader.getValueClass().equals(valueClass)) {
      throw new Exception(dr + "Wrong value class in " + dir + ", expected" + valueClass.getName() +
                          ", got " + dataReader.getValueClass().getName());
    }
    long cnt = 0L;
    Writable key = ReflectionUtils.newInstance(keyClass, conf);
    Writable value = ReflectionUtils.newInstance(valueClass, conf);
    SequenceFile.Writer indexWriter = null;
    if (!dryrun) {
      indexWriter = 
        SequenceFile.createWriter(conf, 
                                  SequenceFile.Writer.file(index), 
                                  SequenceFile.Writer.keyClass(keyClass), 
                                  SequenceFile.Writer.valueClass
                                    (LongWritable.class));
    }
    try {
      /** What's the position (in bytes) we wrote when we got the last index */
      long lastIndexPos = -1;
      /**
       * What was size when we last wrote an index. Set to MIN_VALUE to ensure
       * that we have an index at position zero - midKey will throw an exception
       * if this is not the case
       */
      long lastIndexKeyCount = Long.MIN_VALUE;
      long pos = dataReader.getPosition();
      LongWritable position = new LongWritable();
      long nextBlock = pos;
      boolean blockCompressed = dataReader.isBlockCompressed();
      while(dataReader.next(key, value)) {
        if (blockCompressed) {
          long curPos = dataReader.getPosition();
          if (curPos > nextBlock) {
            pos = nextBlock;                       // current block position
            nextBlock = curPos;
          }
        }
        // Follow the same logic as in
        // {@link MapFile.Writer#append(WritableComparable, Writable)}
        if (cnt >= lastIndexKeyCount + indexInterval && pos > lastIndexPos) {
          position.set(pos);
          if (!dryrun) {
            indexWriter.append(key, position);
          }
          lastIndexPos = pos;
          lastIndexKeyCount = cnt;
        }
        if (!blockCompressed) {
          pos = dataReader.getPosition();         // next record position
        }
        cnt++;
      }
    } catch(Throwable t) {
      // truncated data file. swallow it.
    }
    dataReader.close();
    if (!dryrun) indexWriter.close();
    return cnt;
  }
@Test
  public void testFix() {
    final String INDEX_LESS_MAP_FILE = "testFix.mapfile";
    int PAIR_SIZE = 20;
    MapFile.Writer writer = null;
    try {
      FileSystem fs = FileSystem.getLocal(conf);
      Path dir = new Path(TEST_DIR, INDEX_LESS_MAP_FILE);
      writer = createWriter(INDEX_LESS_MAP_FILE, IntWritable.class, Text.class);
      for (int i = 0; i < PAIR_SIZE; i++)
        writer.append(new IntWritable(0), new Text("value"));
      writer.close();

      File indexFile = new File(".", "." + INDEX_LESS_MAP_FILE + "/index");
      boolean isDeleted = false;
      if (indexFile.exists())
        isDeleted = indexFile.delete();

      if (isDeleted)
        assertTrue("testFix error !!!",
            MapFile.fix(fs, dir, IntWritable.class, Text.class, true, conf) == PAIR_SIZE);
    } catch (Exception ex) {
      fail("testFix error !!!");
    } finally {
      IOUtils.cleanupWithLogger(LOG, writer);
    }
  }