YARN-9791. Queue Mutation API does not allow to remove a config. Contributed by Prabhu Joseph.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/MutableCSConfigurationProvider.java

@@ -280,12 +280,14 @@ public class MutableCSConfigurationProvider implements CSConfigurationProvider,
       String keyPrefix = CapacitySchedulerConfiguration.PREFIX
           + queuePath + CapacitySchedulerConfiguration.DOT;
       for (Map.Entry<String, String> kv : updateInfo.getParams().entrySet()) {
-        if (kv.getValue() == null) {
+        String keyValue = kv.getValue();
+        if (keyValue == null || keyValue.isEmpty()) {
+          keyValue = null;
           proposedConf.unset(keyPrefix + kv.getKey());
         } else {
-          proposedConf.set(keyPrefix + kv.getKey(), kv.getValue());
+          proposedConf.set(keyPrefix + kv.getKey(), keyValue);
         }
-        confUpdate.put(keyPrefix + kv.getKey(), kv.getValue());
+        confUpdate.put(keyPrefix + kv.getKey(), keyValue);
       }
     }
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestMutableCSConfigurationProvider.java

@@ -106,6 +106,43 @@ public class TestMutableCSConfigurationProvider {
         "yarn.scheduler.capacity.root.a.badKey"));
   }
 
+  @Test
+  public void testRemoveQueueConfig() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
+        YarnConfiguration.MEMORY_CONFIGURATION_STORE);
+    confProvider.init(conf);
+
+    SchedConfUpdateInfo updateInfo = new SchedConfUpdateInfo();
+    Map<String, String> updateMap = new HashMap<>();
+    updateMap.put("testkey1", "testval1");
+    updateMap.put("testkey2", "testval2");
+    QueueConfigInfo queueConfigInfo = new
+        QueueConfigInfo("root.a", updateMap);
+    updateInfo.getUpdateQueueInfo().add(queueConfigInfo);
+
+    confProvider.logAndApplyMutation(TEST_USER, updateInfo);
+    confProvider.confirmPendingMutation(true);
+    assertEquals("testval1", confProvider.loadConfiguration(conf)
+        .get("yarn.scheduler.capacity.root.a.testkey1"));
+    assertEquals("testval2", confProvider.loadConfiguration(conf)
+        .get("yarn.scheduler.capacity.root.a.testkey2"));
+
+    // Unset testkey1.
+    updateInfo = new SchedConfUpdateInfo();
+    updateMap.put("testkey1", "");
+    queueConfigInfo = new QueueConfigInfo("root.a", updateMap);
+    updateInfo.getUpdateQueueInfo().add(queueConfigInfo);
+
+    confProvider.logAndApplyMutation(TEST_USER, updateInfo);
+    confProvider.confirmPendingMutation(true);
+    assertNull("Failed to remove config",
+        confProvider.loadConfiguration(conf)
+        .get("yarn.scheduler.capacity.root.a.testkey1"));
+    assertEquals("testval2", confProvider.loadConfiguration(conf)
+        .get("yarn.scheduler.capacity.root.a.testkey2"));
+  }
+
   @Test
   public void testHDFSBackedProvider() throws Exception {
     File testSchedulerConfigurationDir = new File(