YARN-3634. TestMRTimelineEventHandling and TestApplication are broken. Contributed by Sangjin Lee.

(cherry picked from commit b059dd4882fd759e4762cc11c019be4b68fb74c1)

hadoop-yarn-project/CHANGES.txt

@@ -70,6 +70,9 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1
     YARN-3529. Added mini HBase cluster and Phoenix support to timeline service
     v2 unit tests. (Li Lu via zjshen)
 
+    YARN-3634. TestMRTimelineEventHandling and TestApplication are broken. (
+    Sangjin Lee via junping_du)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/collectormanager/NMCollectorService.java

@@ -81,6 +81,11 @@ public class NMCollectorService extends CompositeService implements
                 YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_THREAD_COUNT));
 
     server.start();
+    collectorServerAddress = conf.updateConnectAddr(
+        YarnConfiguration.NM_BIND_HOST,
+        YarnConfiguration.NM_COLLECTOR_SERVICE_ADDRESS,
+        YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_ADDRESS,
+        server.getListenerAddress());
     // start remaining services
     super.serviceStart();
     LOG.info("NMCollectorService started at " + collectorServerAddress);

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

@@ -933,7 +933,7 @@ public class ContainerManagerImpl extends CompositeService implements
             TimelineUtils.FLOW_RUN_ID_TAG_PREFIX);
         long flowRunId = 0L;
         if (flowRunIdStr != null && !flowRunIdStr.isEmpty()) {
-          flowRunId = Long.valueOf(flowRunIdStr);
+          flowRunId = Long.parseLong(flowRunIdStr);
         }
         Application application = new ApplicationImpl(dispatcher, user,
             flowName, flowVersion, flowRunId, applicationID, credentials, context);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/application/TestApplication.java

@@ -519,7 +519,8 @@ public class TestApplication {
       when(context.getApplicationACLsManager()).thenReturn(
         new ApplicationACLsManager(conf));
       when(context.getNMTokenSecretManager()).thenReturn(nmTokenSecretMgr);
-      
+      when(context.getConf()).thenReturn(conf);
+
       // Setting master key
       MasterKey masterKey = new MasterKeyPBImpl();
       masterKey.setKeyId(123);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java

@@ -69,9 +69,7 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager {
 
   private String timelineRestServerBindAddress;
 
-  private CollectorNodemanagerProtocol nmCollectorService;
-
-  private InetSocketAddress nmCollectorServiceAddress;
+  private volatile CollectorNodemanagerProtocol nmCollectorService;
 
   static final String COLLECTOR_MANAGER_ATTR_KEY = "collector.manager";
 
@@ -84,19 +82,8 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager {
     super(NodeTimelineCollectorManager.class.getName());
   }
 
-  @Override
-  public void serviceInit(Configuration conf) throws Exception {
-    this.nmCollectorServiceAddress = conf.getSocketAddr(
-        YarnConfiguration.NM_BIND_HOST,
-        YarnConfiguration.NM_COLLECTOR_SERVICE_ADDRESS,
-        YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_ADDRESS,
-        YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_PORT);
-    super.serviceInit(conf);
-  }
-
   @Override
   protected void serviceStart() throws Exception {
-    nmCollectorService = getNMCollectorService();
     startWebApp();
     super.serviceStart();
   }
@@ -176,7 +163,7 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager {
             this.timelineRestServerBindAddress);
     LOG.info("Report a new collector for application: " + appId +
         " to the NM Collector Service.");
-    nmCollectorService.reportNewCollectorInfo(request);
+    getNMCollectorService().reportNewCollectorInfo(request);
   }
 
   private void updateTimelineCollectorContext(
@@ -186,7 +173,7 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager {
         GetTimelineCollectorContextRequest.newInstance(appId);
     LOG.info("Get timeline collector context for " + appId);
     GetTimelineCollectorContextResponse response =
-        nmCollectorService.getTimelineCollectorContext(request);
+        getNMCollectorService().getTimelineCollectorContext(request);
     String userId = response.getUserId();
     if (userId != null && !userId.isEmpty()) {
       collector.getTimelineEntityContext().setUserId(userId);
@@ -207,13 +194,26 @@ public class NodeTimelineCollectorManager extends TimelineCollectorManager {
 
   @VisibleForTesting
   protected CollectorNodemanagerProtocol getNMCollectorService() {
-    Configuration conf = getConfig();
-    final YarnRPC rpc = YarnRPC.create(conf);
-
-    // TODO Security settings.
-    return (CollectorNodemanagerProtocol) rpc.getProxy(
-        CollectorNodemanagerProtocol.class,
-        nmCollectorServiceAddress, conf);
+    if (nmCollectorService == null) {
+      synchronized (this) {
+        if (nmCollectorService == null) {
+          Configuration conf = getConfig();
+          InetSocketAddress nmCollectorServiceAddress = conf.getSocketAddr(
+              YarnConfiguration.NM_BIND_HOST,
+              YarnConfiguration.NM_COLLECTOR_SERVICE_ADDRESS,
+              YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_ADDRESS,
+              YarnConfiguration.DEFAULT_NM_COLLECTOR_SERVICE_PORT);
+          LOG.info("nmCollectorServiceAddress: " + nmCollectorServiceAddress);
+          final YarnRPC rpc = YarnRPC.create(conf);
+
+          // TODO Security settings.
+          nmCollectorService = (CollectorNodemanagerProtocol) rpc.getProxy(
+              CollectorNodemanagerProtocol.class,
+              nmCollectorServiceAddress, conf);
+        }
+      }
+    }
+    return nmCollectorService;
   }
 
   @VisibleForTesting