YARN-11211. QueueMetrics leaks Configuration objects when validation API is called multiple times

Author: 9uapaw

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/PartitionQueueMetrics.java

@@ -78,7 +78,7 @@ public synchronized QueueMetrics getUserMetrics(String userName) {
       metrics = new PartitionQueueMetrics(this.metricsSystem, this.queueName,
           null, false, this.conf, this.partition);
       users.put(userName, metrics);
-      metricsSystem.register(
+      registerMetrics(
           pSourceName(partitionJMXStr).append(qSourceName(queueName))
               .append(",user=").append(userName).toString(),
           "Metrics for user '" + userName + "' in queue '" + queueName + "'",

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/QueueMetrics.java

@@ -289,7 +289,7 @@ public synchronized QueueMetrics getUserMetrics(String userName) {
       metrics =
           new QueueMetrics(metricsSystem, queueName, null, false, conf);
       users.put(userName, metrics);
-      metricsSystem.register(
+      registerMetrics(
           sourceName(queueName).append(",user=").append(userName).toString(),
           "Metrics for user '"+ userName +"' in queue '"+ queueName +"'",
           metrics.tag(QUEUE_INFO, queueName).tag(USER_INFO, userName));
@@ -334,7 +334,7 @@ public synchronized QueueMetrics getPartitionQueueMetrics(String partition) {
       QueueMetrics queueMetrics =
           new PartitionQueueMetrics(metricsSystem, this.queueName, parentQueue,
               this.enableUserMetrics, this.conf, partition);
-      metricsSystem.register(
+      registerMetrics(
           pSourceName(partitionJMXStr).append(qSourceName(this.queueName))
               .toString(),
           "Metrics for queue: " + this.queueName,
@@ -378,7 +378,7 @@ private QueueMetrics getPartitionMetrics(String partition) {
 
       // Register with the MetricsSystems
       if (metricsSystem != null) {
-        metricsSystem.register(pSourceName(partitionJMXStr).toString(),
+        registerMetrics(pSourceName(partitionJMXStr).toString(),
             "Metrics for partition: " + partitionJMXStr,
             (PartitionQueueMetrics) metrics.tag(PARTITION_INFO,
                 partitionJMXStr));
@@ -1359,4 +1359,15 @@ public void setParentQueue(Queue parentQueue) {
       }
     }
   }
+
+  protected void registerMetrics(String sourceName, String desc, QueueMetrics metrics) {
+    MetricsSource source = metricsSystem.getSource(sourceName);
+    // Unregister metrics if a source is already present
+    if (source != null) {
+      LOG.info("Unregistering source " + sourceName);
+      metricsSystem.unregisterSource(sourceName);
+    }
+
+    metricsSystem.register(sourceName, desc, metrics);
+  }
 }

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

@@ -21,6 +21,7 @@
 import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.MetricsSource;
 import org.apache.hadoop.metrics2.MetricsSystem;
 import org.apache.hadoop.metrics2.annotation.Metric;
 import org.apache.hadoop.metrics2.annotation.Metrics;
@@ -240,6 +241,10 @@ public synchronized static CSQueueMetrics forQueue(String queueName,
 
       // Register with the MetricsSystems
       if (ms != null) {
+        MetricsSource source = ms.getSource(sourceName(queueName).toString());
+        if (source != null) {
+          ms.unregisterSource(sourceName(queueName).toString());
+        }
         metrics =
             ms.register(sourceName(queueName).toString(), "Metrics for queue: "
                 + queueName, metrics);
@@ -260,7 +265,7 @@ public synchronized QueueMetrics getUserMetrics(String userName) {
       metrics =
         new CSQueueMetrics(metricsSystem, queueName, null, false, conf);
       users.put(userName, metrics);
-      metricsSystem.register(
+      registerMetrics(
           sourceName(queueName).append(",user=").append(userName).toString(),
           "Metrics for user '" + userName + "' in queue '" + queueName + "'",
           ((CSQueueMetrics) metrics.tag(QUEUE_INFO, queueName)).tag(USER_INFO,

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

@@ -23,6 +23,7 @@
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,6 +56,8 @@ public static boolean validateCSConfiguration(
       return true;
     } finally {
       newCs.stop();
+      QueueMetrics.clearQueueMetrics();
+      liveScheduler.resetSchedulerMetrics();
     }
   }