YARN-10497. Fix an issue in CapacityScheduler which fails to delete queues. Contributed by Wangda Tan and Qi Zhu.

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

@@ -68,10 +68,18 @@ public class MutableCSConfigurationProvider implements CSConfigurationProvider,
     this.rmContext = rmContext;
   }
 
+  // Unit test can overwrite this method
+  protected Configuration getInitSchedulerConfig() {
+    Configuration initialSchedConf = new Configuration(false);
+    initialSchedConf.
+        addResource(YarnConfiguration.CS_CONFIGURATION_FILE);
+    return initialSchedConf;
+  }
+
   @Override
   public void init(Configuration config) throws IOException {
     this.confStore = YarnConfigurationStoreFactory.getStore(config);
-    Configuration initialSchedConf = new Configuration(false);
+    Configuration initialSchedConf = getInitSchedulerConfig();
     initialSchedConf.addResource(YarnConfiguration.CS_CONFIGURATION_FILE);
     this.schedConf = new Configuration(false);
     // We need to explicitly set the key-values in schedConf, otherwise
@@ -231,7 +239,7 @@ public class MutableCSConfigurationProvider implements CSConfigurationProvider,
     String childQueuesKey = CapacitySchedulerConfiguration.PREFIX +
         parentQueue + CapacitySchedulerConfiguration.DOT +
         CapacitySchedulerConfiguration.QUEUES;
-    return new ArrayList<>(conf.getStringCollection(childQueuesKey));
+    return new ArrayList<>(conf.getTrimmedStringCollection(childQueuesKey));
   }
 
   private Map<String, String> constructKeyValueConfUpdate(

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

@@ -217,6 +217,39 @@ public class TestMutableCSConfigurationProvider {
 
   }
 
+  @Test
+  public void testAddRemoveQueueWithSpacesInConfig() throws Exception {
+    CapacitySchedulerConfiguration csConf =
+        new CapacitySchedulerConfiguration();
+    csConf.setQueues(CapacitySchedulerConfiguration.ROOT,
+        new String[] {" a   , b, c" });
+
+    final String a = CapacitySchedulerConfiguration.ROOT + ".a";
+    final String b = CapacitySchedulerConfiguration.ROOT + ".b";
+    final String c = CapacitySchedulerConfiguration.ROOT + ".c";
+    csConf.setCapacity(a, 0);
+    csConf.setCapacity(b, 50);
+    csConf.setCapacity(c, 50);
+
+    confProvider = new MutableCSConfigurationProvider(rmContext) {
+      @Override
+      protected Configuration getInitSchedulerConfig() {
+        return csConf;
+      }
+    };
+
+    Configuration conf = new Configuration();
+    conf.set(YarnConfiguration.SCHEDULER_CONFIGURATION_STORE_CLASS,
+        YarnConfiguration.MEMORY_CONFIGURATION_STORE);
+    confProvider.init(conf);
+
+    SchedConfUpdateInfo update = new SchedConfUpdateInfo();
+    update.getRemoveQueueInfo().add("root.a");
+
+    confProvider.logAndApplyMutation(UserGroupInformation
+        .getCurrentUser(), update);
+  }
+
   private void writeConf(Configuration conf, String storePath)
       throws IOException {
     FileSystem fileSystem = FileSystem.get(new Configuration(conf));