YARN-4795. ContainerMetrics drops records. (Daniel Templeton via kasha)

(cherry picked from commit 1a3f1482e2738c7f9a983bc55189116930388d7b)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/ContainerMetrics.java

@@ -121,11 +121,10 @@ public class ContainerMetrics implements MetricsSource {
   final MetricsSystem metricsSystem;
 
   // Metrics publishing status
-  private long flushPeriodMs;
+  private final long flushPeriodMs;
+  private final long unregisterDelayMs;
   private boolean flushOnPeriod = false; // true if period elapsed
   private boolean finished = false; // true if container finished
-  private boolean unregister = false; // unregister
-  private long unregisterDelayMs;
   private Timer timer; // lazily initialized
 
   /**
@@ -227,17 +226,11 @@ public class ContainerMetrics implements MetricsSource {
   @Override
   public synchronized void getMetrics(MetricsCollector collector, boolean all) {
     //Container goes through registered -> finished -> unregistered.
-    if (unregister) {
-      return;
-    }
-
     if (finished || flushOnPeriod) {
       registry.snapshot(collector.addRecord(registry.info()), all);
     }
 
-    if (finished) {
-      this.unregister = true;
-    } else if (flushOnPeriod) {
+    if (!finished && flushOnPeriod) {
       flushOnPeriod = false;
       scheduleTimerTaskIfRequired();
     }
@@ -301,6 +294,7 @@ public class ContainerMetrics implements MetricsSource {
       if (timer == null) {
         this.timer = new Timer("Metrics flush checker", true);
       }
+
       TimerTask timerTask = new TimerTask() {
         @Override
         public void run() {
@@ -311,6 +305,7 @@ public class ContainerMetrics implements MetricsSource {
           }
         }
       };
+
       timer.schedule(timerTask, flushPeriodMs);
     }
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/monitor/TestContainerMetrics.java

@@ -77,11 +77,12 @@ public class TestContainerMetrics {
     collector.clear();
 
     metrics.getMetrics(collector, true);
-    assertEquals(ERR, 0, collector.getRecords().size());
+    assertEquals(ERR, 1, collector.getRecords().size());
+    collector.clear();
 
     Thread.sleep(110);
     metrics.getMetrics(collector, true);
-    assertEquals(ERR, 0, collector.getRecords().size());
+    assertEquals(ERR, 1, collector.getRecords().size());
   }
 
   @Test
@@ -134,7 +135,7 @@ public class TestContainerMetrics {
   public void testContainerMetricsFinished() throws InterruptedException {
     MetricsSystemImpl system = new MetricsSystemImpl();
     system.init("test");
-    MetricsCollectorImpl collector = new MetricsCollectorImpl();
+
     ApplicationId appId = ApplicationId.newInstance(1234, 3);
     ApplicationAttemptId appAttemptId =
         ApplicationAttemptId.newInstance(appId, 4);