HADOOP-17081. MetricsSystem doesn't start the sink adapters on restart (#2089)

Contributed by Madhusoodan P

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/metrics2/impl/MetricsSystemImpl.java

@@ -273,7 +273,11 @@ public class MetricsSystemImpl extends MetricsSystem implements MetricsSource {
   T register(final String name, final String description, final T sink) {
     LOG.debug(name +", "+ description);
     if (allSinks.containsKey(name)) {
-      LOG.warn("Sink "+ name +" already exists!");
+      if(sinks.get(name) == null) {
+        registerSink(name, description, sink);
+      } else {
+        LOG.warn("Sink "+ name +" already exists!");
+      }
       return sink;
     }
     allSinks.put(name, sink);

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/metrics2/impl/TestMetricsSystemImpl.java

@@ -639,4 +639,25 @@ public class TestMetricsSystemImpl {
   private static String getPluginUrlsAsString() {
     return "file:metrics2-test-plugin.jar";
   }
+
+  @Test
+  public void testMetricSystemRestart() {
+    MetricsSystemImpl ms = new MetricsSystemImpl("msRestartTestSystem");
+    TestSink ts = new TestSink();
+    String sinkName = "restartTestSink";
+
+    try {
+      ms.start();
+      ms.register(sinkName, "", ts);
+      assertNotNull("no adapter exists for " + sinkName,
+              ms.getSinkAdapter(sinkName));
+      ms.stop();
+
+      ms.start();
+      assertNotNull("no adapter exists for " + sinkName,
+              ms.getSinkAdapter(sinkName));
+    } finally {
+      ms.stop();
+    }
+  }
 }