@VisibleForTesting
  protected static MapOutputFile renameMapOutputForReduce(JobConf conf,
      TaskAttemptId mapId, MapOutputFile subMapOutputFile) throws IOException {
    FileSystem localFs = FileSystem.getLocal(conf);
    // move map output to reduce input
    Path mapOut = subMapOutputFile.getOutputFile();
    FileStatus mStatus = localFs.getFileStatus(mapOut);
    Path reduceIn = subMapOutputFile.getInputFileForWrite(
        TypeConverter.fromYarn(mapId).getTaskID(), mStatus.getLen());
    Path mapOutIndex = subMapOutputFile.getOutputIndexFile();
    Path reduceInIndex = new Path(reduceIn.toString() + ".index");
    if (LOG.isDebugEnabled()) {
      LOG.debug("Renaming map output file for task attempt "
          + mapId.toString() + " from original location " + mapOut.toString()
          + " to destination " + reduceIn.toString());
    }
    if (!localFs.mkdirs(reduceIn.getParent())) {
      throw new IOException("Mkdirs failed to create "
          + reduceIn.getParent().toString());
    }
    if (!localFs.rename(mapOut, reduceIn))
      throw new IOException("Couldn't rename " + mapOut);
    if (!localFs.rename(mapOutIndex, reduceInIndex))
      throw new IOException("Couldn't rename " + mapOutIndex);

    return new RenamedMapOutputFile(reduceIn);
  }
@Test
  public void testRenameMapOutputForReduce() throws Exception {
    final JobConf conf = new JobConf();

    final MROutputFiles mrOutputFiles = new MROutputFiles();
    mrOutputFiles.setConf(conf);

    // make sure both dirs are distinct
    //
    conf.set(MRConfig.LOCAL_DIR, localDirs[0].toString());
    final Path mapOut = mrOutputFiles.getOutputFileForWrite(1);
    conf.set(MRConfig.LOCAL_DIR, localDirs[1].toString());
    final Path mapOutIdx = mrOutputFiles.getOutputIndexFileForWrite(1);
    Assert.assertNotEquals("Paths must be different!",
        mapOut.getParent(), mapOutIdx.getParent());

    // make both dirs part of LOCAL_DIR
    conf.setStrings(MRConfig.LOCAL_DIR, localDirs);

    final FileContext lfc = FileContext.getLocalFSFileContext(conf);
    lfc.create(mapOut, EnumSet.of(CREATE)).close();
    lfc.create(mapOutIdx, EnumSet.of(CREATE)).close();

    final JobId jobId = MRBuilderUtils.newJobId(12345L, 1, 2);
    final TaskId tid = MRBuilderUtils.newTaskId(jobId, 0, TaskType.MAP);
    final TaskAttemptId taid = MRBuilderUtils.newTaskAttemptId(tid, 0);

    LocalContainerLauncher.renameMapOutputForReduce(conf, taid, mrOutputFiles);
  }