From d275677e240cca469a4cec3dc7238e3f18b3f7e9 Mon Sep 17 00:00:00 2001 From: Junping Du Date: Wed, 13 May 2015 11:54:24 -0700 Subject: [PATCH] YARN-3634. TestMRTimelineEventHandling and TestApplication are broken. Contributed by Sangjin Lee. (cherry picked from commit b059dd4882fd759e4762cc11c019be4b68fb74c1) --- hadoop-project/pom.xml | 12 +++++ .../collectormanager/NMCollectorService.java | 5 +++ .../ContainerManagerImpl.java | 2 +- .../application/TestApplication.java | 3 +- .../NodeTimelineCollectorManager.java | 44 +++++++++---------- 5 files changed, 42 insertions(+), 24 deletions(-) diff --git a/hadoop-project/pom.xml b/hadoop-project/pom.xml index 12043e25324..484d22ddc67 100644 --- a/hadoop-project/pom.xml +++ b/hadoop-project/pom.xml @@ -1053,6 +1053,13 @@ org.apache.hbase hbase-client ${hbase.version} + + + + jdk.tools + jdk.tools + + org.apache.phoenix @@ -1091,6 +1098,11 @@ test true + + + jdk.tools + jdk.tools + org.jruby jruby-complete diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/collectormanager/NMCollectorService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/collectormanager/NMCollectorService.java index dc5601ffd2d..db79ee5bc22 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/collectormanager/NMCollectorService.java +++ b/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); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java index 722f5fc8772..6fd1f550dc8 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java @@ -962,7 +962,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); diff --git a/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 b/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 index a91e4b0e17e..dc599e7d597 100644 --- a/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 +++ b/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 @@ -539,7 +539,8 @@ public class TestApplication { new ApplicationACLsManager(conf)); when(context.getNMTokenSecretManager()).thenReturn(nmTokenSecretMgr); when(context.getNMStateStore()).thenReturn(stateStoreService); - + when(context.getConf()).thenReturn(conf); + // Setting master key MasterKey masterKey = new MasterKeyPBImpl(); masterKey.setKeyId(123); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java index 03ac7708651..31051db1687 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/collector/NodeTimelineCollectorManager.java +++ b/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); + 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. - return (CollectorNodemanagerProtocol) rpc.getProxy( - CollectorNodemanagerProtocol.class, - nmCollectorServiceAddress, conf); + // TODO Security settings. + nmCollectorService = (CollectorNodemanagerProtocol) rpc.getProxy( + CollectorNodemanagerProtocol.class, + nmCollectorServiceAddress, conf); + } + } + } + return nmCollectorService; } @VisibleForTesting