HIVE-29667: [hiveACIDRepl] User is unable to assign yarn queue name to acid policies effectively via UI

Author: Rajesh Sarkar

shims/0.23/src/main/java/org/apache/hadoop/hive/shims/Hadoop23Shims.java

@@ -119,6 +119,8 @@ public class Hadoop23Shims extends HadoopShimsSecure {
 
   HadoopShims.MiniDFSShim cluster = null;
   final boolean storagePolicy;
+  private final String DEFAULT = "default";
+  private final String MAPRED_JOB_QUEUE_NAME = "mapred.job.queue.name";
 
   public Hadoop23Shims() {
     // in-memory HDFS
@@ -1238,6 +1240,13 @@ protected int runDistCpInternal(DistCp distcp, List<String> params) {
    * of these details.
    */
   protected void ensureMapReduceQueue(Configuration conf) {
+    String mapredQueue = conf.getRaw(MAPRED_JOB_QUEUE_NAME);
+    if (StringUtils.isNotEmpty(mapredQueue) && !DEFAULT.equals(mapredQueue)) {
+      LOG.info("DistCp: setting mapreduce.job.queuename to '{}' from mapred.job.queue.name", mapredQueue);
+      conf.set(MRJobConfig.QUEUE_NAME, mapredQueue);
+      return;
+    }
+
     String queueName = conf.get(TezConfiguration.TEZ_QUEUE_NAME);
     boolean isTez = "tez".equalsIgnoreCase(conf.get("hive.execution.engine"));
     boolean shouldMapredJobsFollowTezQueue = conf.getBoolean("hive.mapred.job.follow.tez.queue", false);

shims/0.23/src/main/test/org/apache/hadoop/hive/shims/TestHadoop23Shims.java

@@ -196,13 +196,22 @@ public void testGetFileIdForNonexistingPath() throws Exception {
   @Test
   public void testMapReduceQueueIsSetToTezQueue() throws Exception {
     Configuration conf = new Configuration();
+    // mapred.job.queue.name set by replication policy
+    conf = new Configuration();
+    conf.set("mapred.job.queue.name", "testqueue");
+    conf.set("hive.execution.engine", "tez");
+    DistCp distCp = runMockDistCp(conf);
+    assertEquals("testqueue", distCp.getConf().get(MRJobConfig.QUEUE_NAME));
+
     // there is a tez.queue.name, but hive.mapred.job.follow.tez.queue is not allowed
+    conf = new Configuration();
     conf.set(TezConfiguration.TEZ_QUEUE_NAME, "helloQ");
     conf.set("hive.execution.engine", "tez");
-    DistCp distCp = runMockDistCp(conf);
+    distCp = runMockDistCp(conf);
     assertEquals("default", distCp.getConf().get(MRJobConfig.QUEUE_NAME));
 
     // there is a tez.queue.name, and hive.mapred.job.follow.tez.queue is allowed
+    conf = new Configuration();
     conf.set(TezConfiguration.TEZ_QUEUE_NAME, "helloQ");
     conf.setBoolean("hive.mapred.job.follow.tez.queue", true);
     conf.set("hive.execution.engine", "tez");