LocalContainerLauncher extends AbstractService implements
    ContainerLauncher { @VisibleForTesting protected static MapOutputFile renameMapOutputForReduce(JobConf conf, TaskAttemptId mapId, MapOutputFile subMapOutputFile) throws IOException { FileSystem localFs = FileSystem.getLocal(conf); 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); } LocalContainerLauncher(AppContext context,
                                TaskUmbilicalProtocol umbilical); void serviceStart(); void serviceStop(); @Override void handle(ContainerLauncherEvent event); void setEncryptedSpillKey(byte[] encryptedSpillKey);  }
@Test public void testRenameMapOutputForReduce() throws Exception { final JobConf conf = new JobConf(); final MROutputFiles mrOutputFiles = new MROutputFiles(); mrOutputFiles.setConf(conf); 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()); 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); }