You are viewing a plain text version of this content. The canonical link for it is here.
Posted to yarn-commits@hadoop.apache.org by vi...@apache.org on 2013/06/21 02:09:46 UTC
svn commit: r1495248 - in
/hadoop/common/branches/branch-2/hadoop-yarn-project: ./
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/
hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/
hadoo...
Author: vinodkv
Date: Fri Jun 21 00:09:45 2013
New Revision: 1495248
URL: http://svn.apache.org/r1495248
Log:
YARN-851. Share NMTokens using NMTokenCache (api-based) between AMRMClient and NMClient instead of memory based approach which is used currently. Contributed by Omkar Vinit Joshi.
svn merge --ignore-ancestry -c 1495247 ../../trunk/
Added:
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMTokenCache.java
- copied unchanged from r1495247, hadoop/common/trunk/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMTokenCache.java
Modified:
hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RegisterApplicationMasterResponse.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMClient.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/NMClientAsync.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/NMClientAsyncImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/ContainerManagementProtocolProxy.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/CHANGES.txt Fri Jun 21 00:09:45 2013
@@ -187,6 +187,10 @@ Release 2.1.0-beta - UNRELEASED
ApplicationSubmissionContext to simplify the api. (Karthik Kambatla via
acmurthy)
+ YARN-851. Share NMTokens using NMTokenCache (api-based) between AMRMClient
+ and NMClient instead of memory based approach which is used currently. (Omkar
+ Vinit Joshi via vinodkv)
+
NEW FEATURES
YARN-482. FS: Extend SchedulingMode to intermediate queues.
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RegisterApplicationMasterResponse.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RegisterApplicationMasterResponse.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RegisterApplicationMasterResponse.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/RegisterApplicationMasterResponse.java Fri Jun 21 00:09:45 2013
@@ -36,8 +36,9 @@ import org.apache.hadoop.yarn.util.Recor
*
* <p>The response contains critical details such as:
* <ul>
- * <li>Minimum capability for allocated resources in the cluster.</li>
* <li>Maximum capability for allocated resources in the cluster.</li>
+ * <li><code>ApplicationACL</code>s for the application.</li>
+ * <li>ClientToAMToken master key.</li>
* </ul>
* </p>
*
@@ -50,11 +51,12 @@ public abstract class RegisterApplicatio
@Unstable
public static RegisterApplicationMasterResponse newInstance(
Resource minCapability, Resource maxCapability,
- Map<ApplicationAccessType, String> acls) {
+ Map<ApplicationAccessType, String> acls, ByteBuffer key) {
RegisterApplicationMasterResponse response =
Records.newRecord(RegisterApplicationMasterResponse.class);
response.setMaximumResourceCapability(maxCapability);
response.setApplicationACLs(acls);
+ response.setClientToAMTokenMasterKey(key);
return response;
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/Container.java Fri Jun 21 00:09:45 2013
@@ -44,12 +44,10 @@ import org.apache.hadoop.yarn.util.Recor
* <li>HTTP uri of the node.</li>
* <li>{@link Resource} allocated to the container.</li>
* <li>{@link Priority} at which the container was allocated.</li>
- * <li>{@link ContainerState} of the container.</li>
* <li>
* Container {@link Token} of the container, used to securely verify
* authenticity of the allocation.
* </li>
- * <li>{@link ContainerStatus} of the container.</li>
* </ul>
* </p>
*
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/main/java/org/apache/hadoop/yarn/applications/distributedshell/ApplicationMaster.java Fri Jun 21 00:09:45 2013
@@ -448,8 +448,7 @@ public class ApplicationMaster {
resourceManager.start();
containerListener = new NMCallbackHandler();
- nmClientAsync =
- new NMClientAsyncImpl(containerListener, resourceManager.getNMTokens());
+ nmClientAsync = new NMClientAsyncImpl(containerListener);
nmClientAsync.init(conf);
nmClientAsync.start();
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/src/test/java/org/apache/hadoop/yarn/applications/distributedshell/TestDistributedShell.java Fri Jun 21 00:09:45 2013
@@ -100,7 +100,7 @@ public class TestDistributedShell {
}
}
- @Test(timeout=30000)
+ @Test(timeout=90000)
public void testDSShell() throws Exception {
String[] args = {
@@ -128,7 +128,7 @@ public class TestDistributedShell {
}
- @Test(timeout=30000)
+ @Test(timeout=90000)
public void testDSShellWithInvalidArgs() throws Exception {
Client client = new Client(new Configuration(yarnCluster.getConfig()));
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java Fri Jun 21 00:09:45 2013
@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.client.ap
import java.io.IOException;
import java.util.Collection;
import java.util.List;
-import java.util.concurrent.ConcurrentMap;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -35,7 +34,6 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.client.api.impl.AMRMClientImpl;
import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -249,14 +247,4 @@ public abstract class AMRMClient<T exten
Priority priority,
String resourceName,
Resource capability);
-
- /**
- * It returns the NMToken received on allocate call. It will not communicate
- * with RM to get NMTokens. On allocate call whenever we receive new token
- * along with container AMRMClient will cache this NMToken per node manager.
- * This map returned should be shared with any application which is
- * communicating with NodeManager (ex. NMClient) using NMTokens. If a new
- * NMToken is received for the same node manager then it will be replaced.
- */
- public abstract ConcurrentMap<String, Token> getNMTokens();
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMClient.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMClient.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/NMClient.java Fri Jun 21 00:09:45 2013
@@ -22,21 +22,17 @@ package org.apache.hadoop.yarn.client.ap
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Map;
-import java.util.concurrent.ConcurrentMap;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.service.AbstractService;
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.NMToken;
import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.client.api.impl.NMClientImpl;
import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -46,30 +42,19 @@ public abstract class NMClient extends A
/**
* Create a new instance of NMClient.
- * @param nmTokens need to pass map of NMTokens which are received on
- * {@link AMRMClient#allocate(float)} call as a part of
- * {@link AllocateResponse}.
- * key :- NodeAddr (host:port)
- * Value :- Token {@link NMToken#getToken()}
*/
@Public
- public static NMClient createNMClient(ConcurrentMap<String, Token> nmTokens) {
- NMClient client = new NMClientImpl(nmTokens);
+ public static NMClient createNMClient() {
+ NMClient client = new NMClientImpl();
return client;
}
/**
* Create a new instance of NMClient.
- * @param nmTokens need to pass map of NMTokens which are received on
- * {@link AMRMClient#allocate(float)} call as a part of
- * {@link AllocateResponse}.
- * key :- NodeAddr (host:port)
- * Value :- Token {@link NMToken#getToken()}
*/
@Public
- public static NMClient createNMClient(String name,
- ConcurrentMap<String, Token> nmTokens) {
- NMClient client = new NMClientImpl(name, nmTokens);
+ public static NMClient createNMClient(String name) {
+ NMClient client = new NMClientImpl(name);
return client;
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java Fri Jun 21 00:09:45 2013
@@ -21,7 +21,6 @@ package org.apache.hadoop.yarn.client.ap
import java.io.IOException;
import java.util.Collection;
import java.util.List;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.hadoop.classification.InterfaceAudience.Private;
@@ -37,7 +36,6 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.async.impl.AMRMClientAsyncImpl;
@@ -198,17 +196,6 @@ extends AbstractService {
*/
public abstract int getClusterNodeCount();
- /**
- * It returns the NMToken received on allocate call. It will not communicate
- * with RM to get NMTokens. On allocate call whenever we receive new token
- * along with new container AMRMClientAsync will cache this NMToken per node
- * manager. This map returned should be shared with any application which is
- * communicating with NodeManager (ex. NMClient / NMClientAsync) using
- * NMTokens. If a new NMToken is received for the same node manager
- * then it will be replaced.
- */
- public abstract ConcurrentMap<String, Token> getNMTokens();
-
public interface CallbackHandler {
/**
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/NMClientAsync.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/NMClientAsync.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/NMClientAsync.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/NMClientAsync.java Fri Jun 21 00:09:45 2013
@@ -112,18 +112,16 @@ public abstract class NMClientAsync exte
protected CallbackHandler callbackHandler;
public static NMClientAsync createNMClientAsync(
- CallbackHandler callbackHandler, ConcurrentMap<String, Token> nmTokens) {
- return new NMClientAsyncImpl(callbackHandler, nmTokens);
+ CallbackHandler callbackHandler) {
+ return new NMClientAsyncImpl(callbackHandler);
}
- protected NMClientAsync(CallbackHandler callbackHandler,
- ConcurrentMap<String, Token> nmTokens) {
- this (NMClientAsync.class.getName(), callbackHandler, nmTokens);
+ protected NMClientAsync(CallbackHandler callbackHandler) {
+ this (NMClientAsync.class.getName(), callbackHandler);
}
- protected NMClientAsync(String name, CallbackHandler callbackHandler,
- ConcurrentMap<String, Token> nmTokens) {
- this (name, new NMClientImpl(nmTokens), callbackHandler);
+ protected NMClientAsync(String name, CallbackHandler callbackHandler) {
+ this (name, new NMClientImpl(), callbackHandler);
}
@Private
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java Fri Jun 21 00:09:45 2013
@@ -22,7 +22,6 @@ import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.commons.logging.Log;
@@ -40,7 +39,6 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
@@ -215,19 +213,6 @@ extends AMRMClientAsync<T> {
public int getClusterNodeCount() {
return client.getClusterNodeCount();
}
-
- /**
- * It returns the NMToken received on allocate call. It will not communicate
- * with RM to get NMTokens. On allocate call whenever we receive new token
- * along with new container AMRMClientAsync will cache this NMToken per node
- * manager. This map returned should be shared with any application which is
- * communicating with NodeManager (ex. NMClient / NMClientAsync) using
- * NMTokens. If a new NMToken is received for the same node manager
- * then it will be replaced.
- */
- public ConcurrentMap<String, Token> getNMTokens() {
- return client.getNMTokens();
- }
private class HeartbeatThread extends Thread {
public HeartbeatThread() {
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/NMClientAsyncImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/NMClientAsyncImpl.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/NMClientAsyncImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/NMClientAsyncImpl.java Fri Jun 21 00:09:45 2013
@@ -82,14 +82,12 @@ public class NMClientAsyncImpl extends N
protected ConcurrentMap<ContainerId, StatefulContainer> containers =
new ConcurrentHashMap<ContainerId, StatefulContainer>();
- public NMClientAsyncImpl(CallbackHandler callbackHandler,
- ConcurrentMap<String, Token> nmTokens) {
- this(NMClientAsync.class.getName(), callbackHandler, nmTokens);
+ public NMClientAsyncImpl(CallbackHandler callbackHandler) {
+ this(NMClientAsync.class.getName(), callbackHandler);
}
- public NMClientAsyncImpl(String name, CallbackHandler callbackHandler,
- ConcurrentMap<String, Token> nmTokens) {
- this(name, new NMClientImpl(nmTokens), callbackHandler);
+ public NMClientAsyncImpl(String name, CallbackHandler callbackHandler) {
+ this(name, new NMClientImpl(), callbackHandler);
}
@Private
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java Fri Jun 21 00:09:45 2013
@@ -34,7 +34,6 @@ import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -56,8 +55,8 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceRequest;
-import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.apache.hadoop.yarn.client.api.NMTokenCache;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -82,7 +81,6 @@ public class AMRMClientImpl<T extends Co
RecordFactoryProvider.getRecordFactory(null);
private int lastResponseId = 0;
- private ConcurrentHashMap<String, Token> nmTokens;
protected ApplicationMasterProtocol rmClient;
protected final ApplicationAttemptId appAttemptId;
@@ -158,7 +156,6 @@ public class AMRMClientImpl<T extends Co
public AMRMClientImpl(ApplicationAttemptId appAttemptId) {
super(AMRMClientImpl.class.getName());
this.appAttemptId = appAttemptId;
- this.nmTokens = new ConcurrentHashMap<String, Token>();
}
@Override
@@ -285,12 +282,12 @@ public class AMRMClientImpl<T extends Co
protected void populateNMTokens(AllocateResponse allocateResponse) {
for (NMToken token : allocateResponse.getNMTokens()) {
String nodeId = token.getNodeId().toString();
- if (nmTokens.containsKey(nodeId)) {
+ if (NMTokenCache.containsNMToken(nodeId)) {
LOG.debug("Replacing token for : " + nodeId);
} else {
LOG.debug("Received new token for : " + nodeId);
}
- nmTokens.put(nodeId, token.getToken());
+ NMTokenCache.setNMToken(nodeId, token.getToken());
}
}
@@ -577,9 +574,4 @@ public class AMRMClientImpl<T extends Co
+ " #asks=" + ask.size());
}
}
-
- @Override
- public ConcurrentHashMap<String, Token> getNMTokens() {
- return nmTokens;
- }
}
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/ContainerManagementProtocolProxy.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/ContainerManagementProtocolProxy.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/ContainerManagementProtocolProxy.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/ContainerManagementProtocolProxy.java Fri Jun 21 00:09:45 2013
@@ -23,7 +23,6 @@ import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
-import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -36,6 +35,7 @@ import org.apache.hadoop.security.token.
import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.Token;
+import org.apache.hadoop.yarn.client.api.NMTokenCache;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.hadoop.yarn.ipc.YarnRPC;
@@ -54,13 +54,10 @@ public class ContainerManagementProtocol
private final int maxConnectedNMs;
private final LinkedHashMap<String, ContainerManagementProtocolProxyData> cmProxy;
- private Map<String, Token> nmTokens;
private final Configuration conf;
private final YarnRPC rpc;
- public ContainerManagementProtocolProxy(Configuration conf,
- Map<String, Token> nmTokens) {
- this.nmTokens = nmTokens;
+ public ContainerManagementProtocolProxy(Configuration conf) {
this.conf = conf;
maxConnectedNMs =
@@ -86,10 +83,10 @@ public class ContainerManagementProtocol
// This get call will update the map which is working as LRU cache.
ContainerManagementProtocolProxyData proxy =
cmProxy.get(containerManagerBindAddr);
-
+
while (proxy != null
&& !proxy.token.getIdentifier().equals(
- nmTokens.get(containerManagerBindAddr).getIdentifier())) {
+ NMTokenCache.getNMToken(containerManagerBindAddr).getIdentifier())) {
LOG.info("Refreshing proxy as NMToken got updated for node : "
+ containerManagerBindAddr);
// Token is updated. check if anyone has already tried closing it.
@@ -112,7 +109,7 @@ public class ContainerManagementProtocol
if (proxy == null) {
proxy =
new ContainerManagementProtocolProxyData(rpc, containerManagerBindAddr,
- containerId, nmTokens.get(containerManagerBindAddr));
+ containerId, NMTokenCache.getNMToken(containerManagerBindAddr));
if (cmProxy.size() > maxConnectedNMs) {
// Number of existing proxy exceed the limit.
String cmAddr = cmProxy.keySet().iterator().next();
@@ -172,10 +169,6 @@ public class ContainerManagementProtocol
cmProxy.clear();
}
- public synchronized void setNMTokens(Map<String, Token> nmTokens) {
- this.nmTokens = nmTokens;
- }
-
public class ContainerManagementProtocolProxyData {
private final String containerManagerBindAddr;
private final ContainerManagementProtocol proxy;
@@ -201,10 +194,12 @@ public class ContainerManagementProtocol
protected ContainerManagementProtocol newProxy(final YarnRPC rpc,
String containerManagerBindAddr, ContainerId containerId, Token token)
throws InvalidToken {
+
if (token == null) {
throw new InvalidToken("No NMToken sent for "
+ containerManagerBindAddr);
}
+
final InetSocketAddress cmAddr =
NetUtils.createSocketAddr(containerManagerBindAddr);
LOG.info("Opening proxy : " + containerManagerBindAddr);
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/NMClientImpl.java Fri Jun 21 00:09:45 2013
@@ -81,18 +81,15 @@ public class NMClientImpl extends NMClie
new ConcurrentHashMap<ContainerId, StartedContainer>();
//enabled by default
- private final AtomicBoolean cleanupRunningContainers = new AtomicBoolean(true);
- private ContainerManagementProtocolProxy cmProxy;
- private ConcurrentMap<String, Token> nmTokens;
+ private final AtomicBoolean cleanupRunningContainers = new AtomicBoolean(true);
+ private ContainerManagementProtocolProxy cmProxy;
- public NMClientImpl(ConcurrentMap<String, Token> nmTokens) {
+ public NMClientImpl() {
super(NMClientImpl.class.getName());
- this.nmTokens = nmTokens;
}
- public NMClientImpl(String name, ConcurrentMap<String, Token> nmTokens) {
+ public NMClientImpl(String name) {
super(name);
- this.nmTokens = nmTokens;
}
@Override
@@ -126,8 +123,7 @@ public class NMClientImpl extends NMClie
@Override
protected void serviceInit(Configuration conf) throws Exception {
super.serviceInit(conf);
- cmProxy =
- new ContainerManagementProtocolProxy(conf, nmTokens);
+ cmProxy = new ContainerManagementProtocolProxy(conf);
}
@Override
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java Fri Jun 21 00:09:45 2013
@@ -26,11 +26,9 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.Collection;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
-import java.util.concurrent.ConcurrentHashMap;
import junit.framework.Assert;
@@ -50,6 +48,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
+import org.apache.hadoop.yarn.api.records.NMToken;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
@@ -57,6 +56,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.Token;
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.AMRMClient;
+import org.apache.hadoop.yarn.client.api.NMTokenCache;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.AMRMClient.StoredContainerRequest;
@@ -488,8 +488,8 @@ public class TestAMRMClient {
int iterationsLeft = 2;
Set<ContainerId> releases = new TreeSet<ContainerId>();
- ConcurrentHashMap<String, Token> nmTokens = amClient.getNMTokens();
- Assert.assertEquals(0, nmTokens.size());
+ NMTokenCache.clearCache();
+ Assert.assertEquals(0, NMTokenCache.numberOfNMTokensInCache());
HashMap<String, Token> receivedNMTokens = new HashMap<String, Token>();
while (allocatedContainerCount < containersRequestedAny
@@ -505,19 +505,13 @@ public class TestAMRMClient {
releases.add(rejectContainerId);
amClient.releaseAssignedContainer(rejectContainerId);
}
- Assert.assertEquals(nmTokens.size(), amClient.getNMTokens().size());
- Iterator<String> nodeI = nmTokens.keySet().iterator();
- while (nodeI.hasNext()) {
- String nodeId = nodeI.next();
- if (!receivedNMTokens.containsKey(nodeId)) {
- receivedNMTokens.put(nodeId, nmTokens.get(nodeId));
- } else {
- Assert.fail("Received token again for : " + nodeId);
+
+ for (NMToken token : allocResponse.getNMTokens()) {
+ String nodeID = token.getNodeId().toString();
+ if (receivedNMTokens.containsKey(nodeID)) {
+ Assert.fail("Received token again for : " + nodeID);
}
- }
- nodeI = receivedNMTokens.keySet().iterator();
- while (nodeI.hasNext()) {
- nmTokens.remove(nodeI.next());
+ receivedNMTokens.put(nodeID, token.getToken());
}
if(allocatedContainerCount < containersRequestedAny) {
@@ -526,7 +520,6 @@ public class TestAMRMClient {
}
}
- Assert.assertEquals(0, amClient.getNMTokens().size());
// Should receive atleast 1 token
Assert.assertTrue(receivedNMTokens.size() > 0
&& receivedNMTokens.size() <= nodeCount);
Modified: hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java?rev=1495248&r1=1495247&r2=1495248&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java (original)
+++ hadoop/common/branches/branch-2/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestNMClient.java Fri Jun 21 00:09:45 2013
@@ -55,6 +55,7 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.NMClient;
+import org.apache.hadoop.yarn.client.api.NMTokenCache;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -74,7 +75,6 @@ public class TestNMClient {
List<NodeReport> nodeReports = null;
ApplicationAttemptId attemptId = null;
int nodeCount = 3;
- ConcurrentHashMap<String, Token> nmTokens;
@Before
public void setup() throws YarnException, IOException {
@@ -136,7 +136,6 @@ public class TestNMClient {
if (iterationsLeft == 0) {
fail("Application hasn't bee started");
}
- nmTokens = new ConcurrentHashMap<String, Token>();
// start am rm client
rmClient =
@@ -148,7 +147,7 @@ public class TestNMClient {
assertEquals(STATE.STARTED, rmClient.getServiceState());
// start am nm client
- nmClient = (NMClientImpl) NMClient.createNMClient(nmTokens);
+ nmClient = (NMClientImpl) NMClient.createNMClient();
nmClient.init(conf);
nmClient.start();
assertNotNull(nmClient);
@@ -173,7 +172,7 @@ public class TestNMClient {
nmClient.stop();
}
- @Test (timeout = 60000)
+ @Test (timeout = 180000)
public void testNMClientNoCleanupOnStop()
throws YarnException, IOException {
@@ -241,7 +240,8 @@ public class TestNMClient {
}
if (!allocResponse.getNMTokens().isEmpty()) {
for (NMToken token : allocResponse.getNMTokens()) {
- nmTokens.put(token.getNodeId().toString(), token.getToken());
+ NMTokenCache.setNMToken(token.getNodeId().toString(),
+ token.getToken());
}
}
if(allocatedContainerCount < containersRequestedAny) {