MAPREDUCE-7131. Job History Server has race condition where it moves files from intermediate to finished but thinks file is in intermediate. Contributed by Anthony Hsu

(cherry picked from commit eb0b5a844f960017f6f48d746174d0f5826f0e5f)

Conflicts:
	hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java

hadoop-mapreduce-project/CHANGES.txt

@@ -1,5 +1,21 @@
 Hadoop MapReduce Change Log
 
+Release 2.7.8 - UNRELEASED
+
+  INCOMPATIBLE CHANGES
+
+  NEW FEATURES
+
+  IMPROVEMENTS
+
+  OPTIMIZATIONS
+
+  BUG FIXES
+
+    MAPREDUCE-7131. Job History Server has race condition where it moves
+    files from intermediate to finished but thinks file is in
+    intermediate. (Anthony Hsu via jlowe)
+
 Release 2.7.7 - 2018-07-18
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java

@@ -1005,7 +1005,17 @@ public class HistoryFileManager extends AbstractService {
   private void moveToDoneNow(final Path src, final Path target)
       throws IOException {
     LOG.info("Moving " + src.toString() + " to " + target.toString());
-    intermediateDoneDirFc.rename(src, target, Options.Rename.NONE);
+    try {
+      intermediateDoneDirFc.rename(src, target, Options.Rename.NONE);
+    } catch (FileNotFoundException e) {
+      if (doneDirFc.util().exists(target)) {
+        LOG.info("Source file " + src.toString() + " not found, but target "
+            + "file " + target.toString() + " already exists. Move already "
+            + "happened.");
+      } else {
+        throw e;
+      }
+    }
   }
 
   private String getJobSummary(FileContext fc, Path path) throws IOException {

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestHistoryFileManager.java

@@ -25,6 +25,7 @@ import java.io.FileNotFoundException;
 import java.util.UUID;
 import java.util.List;
 
+import org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils;
 import org.junit.Assert;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
@@ -245,6 +246,57 @@ public class TestHistoryFileManager {
     Assert.assertFalse(info.didMoveFail());
   }
 
+  /**
+   * This test sets up a scenario where the history files have already been
+   * moved to the "done" directory (so the "intermediate" directory is empty),
+   * but then moveToDone() is called again on the same history file. It
+   * validates that the second moveToDone() still succeeds rather than throws a
+   * FileNotFoundException.
+   */
+  @Test
+  public void testMoveToDoneAlreadyMovedSucceeds() throws Exception {
+    HistoryFileManagerTest historyFileManager = new HistoryFileManagerTest();
+    long jobTimestamp = 1535436603000L;
+    String job = "job_" + jobTimestamp + "_123456789";
+
+    String intermediateDirectory = "/" + UUID.randomUUID();
+    String doneDirectory = "/" + UUID.randomUUID();
+    Configuration conf = dfsCluster.getConfiguration(0);
+    conf.set(JHAdminConfig.MR_HISTORY_INTERMEDIATE_DONE_DIR,
+        intermediateDirectory);
+    conf.set(JHAdminConfig.MR_HISTORY_DONE_DIR, doneDirectory);
+
+    Path intermediateHistoryFilePath = new Path(intermediateDirectory + "/"
+        + job + ".jhist");
+    Path intermediateConfFilePath = new Path(intermediateDirectory + "/"
+        + job + "_conf.xml");
+    Path doneHistoryFilePath = new Path(doneDirectory + "/"
+        + JobHistoryUtils.timestampDirectoryComponent(jobTimestamp) + "/123456/"
+        + job + ".jhist");
+    Path doneConfFilePath = new Path(doneDirectory + "/"
+        + JobHistoryUtils.timestampDirectoryComponent(jobTimestamp)
+        + "/123456/" + job + "_conf.xml");
+
+    dfsCluster.getFileSystem().createNewFile(doneHistoryFilePath);
+    dfsCluster.getFileSystem().createNewFile(doneConfFilePath);
+
+    historyFileManager.serviceInit(conf);
+
+    JobIndexInfo jobIndexInfo = new JobIndexInfo();
+    jobIndexInfo.setJobId(TypeConverter.toYarn(JobID.forName(job)));
+    jobIndexInfo.setFinishTime(jobTimestamp);
+    HistoryFileInfo info = historyFileManager.getHistoryFileInfo(
+        intermediateHistoryFilePath, intermediateConfFilePath, null,
+        jobIndexInfo, false);
+    info.moveToDone();
+
+    Assert.assertFalse(info.isMovePending());
+    Assert.assertEquals(doneHistoryFilePath.toString(),
+        info.getHistoryFile().toUri().getPath());
+    Assert.assertEquals(doneConfFilePath.toString(),
+        info.getConfFile().toUri().getPath());
+  }
+
   static class HistoryFileManagerTest extends HistoryFileManager {
     public HistoryFileManagerTest() {
       super();