TEZ-4547: Add Tez AM JobID to the JobConf

Some committers require a job-wide UUID to function
correctly. Adding the AM JobID to the JobConf
will allow applications to pass that to
the committers that need it.

Author: vnarayanan

tez-mapreduce/src/main/java/org/apache/tez/mapreduce/committer/MROutputCommitter.java

@@ -119,6 +119,7 @@ public void abortOutput(VertexStatus.State finalState) throws IOException {
         || jobConf.getBoolean("mapred.mapper.new-api", false))  {
       newApiCommitter = true;
     }
+    jobConf.set(MRJobConfig.MR_PARENT_JOB_ID, new org.apache.hadoop.mapred.JobID(String.valueOf(getContext().getApplicationId().getClusterTimestamp()), getContext().getApplicationId().getId()).toString());
     LOG.info("Committer for " + getContext().getVertexName() + ":" + getContext().getOutputName() +
         " using " + (newApiCommitter ? "new" : "old") + "mapred API");
 

tez-mapreduce/src/main/java/org/apache/tez/mapreduce/hadoop/MRJobConfig.java

@@ -131,6 +131,8 @@ public interface MRJobConfig {
 
   public static final String CACHE_ARCHIVES_VISIBILITIES = "mapreduce.job.cache.archives.visibilities";
 
+  public static final String MR_PARENT_JOB_ID = "mapreduce.parent.job.id";
+
   public static final String FILEOUTPUTCOMMITTER_ALGORITHM_VERSION = "mapreduce.fileoutputcommitter.algorithm.version";
 
   /**

tez-mapreduce/src/main/java/org/apache/tez/mapreduce/output/MROutput.java

@@ -44,6 +44,7 @@
 import org.apache.hadoop.mapred.FileOutputFormat;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.JobID;
 import org.apache.hadoop.mapred.TaskAttemptID;
 import org.apache.hadoop.mapreduce.OutputCommitter;
 import org.apache.hadoop.mapreduce.OutputFormat;
@@ -417,6 +418,7 @@ protected List<Event> initializeBase() throws IOException, InterruptedException
         .createMockTaskAttemptID(getContext().getApplicationId().getClusterTimestamp(),
             getContext().getTaskVertexIndex(), getContext().getApplicationId().getId(),
             getContext().getTaskIndex(), getContext().getTaskAttemptNumber(), isMapperOutput);
+    jobConf.set(MRJobConfig.MR_PARENT_JOB_ID, new JobID(String.valueOf(getContext().getApplicationId().getClusterTimestamp()), getContext().getApplicationId().getId()).toString());
     jobConf.set(JobContext.TASK_ATTEMPT_ID, taskAttemptId.toString());
     jobConf.set(JobContext.TASK_ID, taskAttemptId.getTaskID().toString());
     jobConf.setBoolean(JobContext.TASK_ISMAP, isMapperOutput);

tez-mapreduce/src/test/java/org/apache/tez/mapreduce/output/TestMROutput.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
 import org.apache.hadoop.mapred.Reporter;
 import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.OutputCommitter;
@@ -57,6 +58,7 @@
 import org.apache.tez.mapreduce.TestUmbilical;
 import org.apache.tez.mapreduce.TezTestUtils;
 import org.apache.tez.mapreduce.hadoop.MRConfig;
+import org.apache.tez.mapreduce.hadoop.MRJobConfig;
 import org.apache.tez.runtime.LogicalIOProcessorRuntimeTask;
 import org.apache.tez.runtime.api.OutputContext;
 import org.apache.tez.runtime.api.ProcessorContext;
@@ -131,6 +133,26 @@ public void testMergeConfig() throws Exception {
     assertEquals("base-value", mergedConf.get("base-key"));
   }
 
+  @Test
+  public void testParentJobIDSet() throws Exception {
+    Configuration conf = new Configuration();
+    conf.setBoolean(MRConfig.IS_MAP_PROCESSOR, true);
+    DataSinkDescriptor dataSink = MROutput
+            .createConfigBuilder(conf, TextOutputFormat.class,
+                    tmpDir.getPath())
+            .build();
+
+    OutputContext outputContext = createMockOutputContext(dataSink.getOutputDescriptor().getUserPayload(),
+            new Configuration(false));
+    MROutput output = new MROutput(outputContext, 2);
+    output.initialize();
+    String invalidJobID = "invalid default";
+    String parentJobID = output.jobConf.get(MRJobConfig.MR_PARENT_JOB_ID, invalidJobID);
+    assertNotEquals(parentJobID,invalidJobID);
+    assertNotEquals(output.jobConf.get(org.apache.hadoop.mapred.JobContext.TASK_ATTEMPT_ID),parentJobID);
+    assertEquals(parentJobID, new JobID(String.valueOf(outputContext.getApplicationId().getClusterTimestamp()),outputContext.getApplicationId().getId()).toString());
+  }
+
   @Test(timeout = 5000)
   public void testOldAPI_TextOutputFormat() throws Exception {
     Configuration conf = new Configuration();