YARN-9396. Fixed duplicated RM Container created event to ATS.

            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/rmcontainer/RMContainerImpl.java

@@ -608,7 +608,10 @@ public class RMContainerImpl implements RMContainer {
       container.eventHandler.handle(new RMAppRunningOnNodeEvent(container
           .getApplicationAttemptId().getApplicationId(), container.nodeId));
 
-      publishNonAMContainerEventstoATS(container);
+      // Opportunistic containers move directly from NEW to ACQUIRED
+      if (container.getState() == RMContainerState.NEW) {
+        publishNonAMContainerEventstoATS(container);
+      }
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/rmcontainer/TestRMContainerImpl.java

@@ -150,7 +150,7 @@ public class TestRMContainerImpl {
         RMContainerEventType.LAUNCHED));
     drainDispatcher.await();
     assertEquals(RMContainerState.RUNNING, rmContainer.getState());
-    verify(publisher, times(2)).containerCreated(any(RMContainer.class),
+    verify(publisher, times(1)).containerCreated(any(RMContainer.class),
         anyLong());
     assertEquals("http://host:3465/node/containerlogs/container_1_0001_01_000001/user",
         rmContainer.getLogURL());
@@ -253,7 +253,7 @@ public class TestRMContainerImpl {
         RMContainerEventType.ACQUIRED));
     drainDispatcher.await();
     assertEquals(RMContainerState.ACQUIRED, rmContainer.getState());
-    verify(publisher, times(2)).containerCreated(any(RMContainer.class),
+    verify(publisher, times(1)).containerCreated(any(RMContainer.class),
         anyLong());
 
     rmContainer.handle(new RMContainerEvent(containerId,
@@ -345,7 +345,7 @@ public class TestRMContainerImpl {
     // RMContainer should be publishing system metrics for all containers.
     // Since there is 1 AM container and 1 non-AM container, there should be 2
     // container created events and 2 container finished events.
-    verify(publisher, times(4)).containerCreated(any(RMContainer.class),
+    verify(publisher, times(2)).containerCreated(any(RMContainer.class),
         anyLong());
     verify(publisher, times(2)).containerFinished(any(RMContainer.class), anyLong());
   }