Merge remote branch 'jvalkeal/SHDP-171'

SHDP-171 Prepare monitor,allocator and launcher.
Tweaks and tidy up for concepts around monitoring, allocation
and launching. This makes code base clean still leaving
monitoring into a state which needs to be radically modify
order to know real monitoring state. This is due
to removal of container states from containers in yarn api.

SHDP-171 Fixes for ClientRmTemplate
Fixing template and re-enabling tests.

SHDP-171 Clean and tweaks for AppmasterCmTemplate

SHDP-171 Remove user and tidy up
Remove user from yarn client because api doesn't use it anymore.
Tidying up classes.

SHDP-171 Fixing some tests for changes in CONTAINER_ID env variable
AM_CONTAINER_ID is now CONTAINER_ID. Changes for this and tests
which failed.

SHDP-171 More changes due to removal of ApplicationAttemptId from api.
Changes to internal apis, handling of connection authentication, etc.

SHDP-171 ApplicationAttemptId not part of FinishApplicationMasterRequest anymore.

SHDP-171 Fixing AbstractBatchAppmaster per changes in ContainerLauncherInterceptor

SHDP-171 Modify ContainerLauncherInterceptor
Adding Container into interceptor because previously it
was part of the ContainerLaunchContext. That's not a case
anymore and we need this info.

SHDP-171 Saving work with first working unit test

SHDP-171 first change set which compiles

Author: jvalkeal

gradle.properties

@@ -22,8 +22,8 @@ hd12HiveVersion = 0.11.0
 hd12PigVersion = 0.11.1
 hd12ThriftVersion = 0.9.0
 
-## Hadoop20 Alpha
-hd20Version = 2.0.6-alpha
+## Hadoop20 GA
+hd20Version = 2.2.0
 hd20HbaseVersion = 0.94.6
 hd20HiveVersion = 0.10.0
 hd20PigVersion = 0.11.1

spring-yarn/spring-yarn-batch/src/main/java/org/springframework/yarn/batch/am/AbstractBatchAppmaster.java

@@ -203,14 +203,14 @@ protected void onContainerCompleted(ContainerStatus status) {
 	}
 
 	@Override
-	public ContainerLaunchContext preLaunch(ContainerLaunchContext context) {
+	public ContainerLaunchContext preLaunch(Container container, ContainerLaunchContext context) {
 		AppmasterService service = getAppmasterService();
 
 		if(log.isDebugEnabled()) {
 			log.debug("Intercept launch context: " + context);
 		}
 
-		StepExecution stepExecution = containerToStepMap.get(context.getContainerId());
+		StepExecution stepExecution = containerToStepMap.get(container.getId());
 		String jobName = remoteStepNames.get(stepExecution);
 
 		if(service != null) {

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/YarnSystemException.java

@@ -15,9 +15,11 @@
  */
 package org.springframework.yarn;
 
+import java.io.IOException;
+
 import org.apache.hadoop.ipc.RemoteException;
-import org.apache.hadoop.yarn.YarnException;
-import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.springframework.dao.UncategorizedDataAccessException;
 
 /**
@@ -42,11 +44,11 @@ public YarnSystemException(YarnException e) {
 	}
 
 	/**
-	 * Constructs YarnSystemException from {@link YarnRemoteException}.
+	 * Constructs YarnSystemException from {@link YarnRuntimeException}.
 	 *
-	 * @param e the {@link YarnRemoteException}
+	 * @param e the {@link YarnRuntimeException}
 	 */
-	public YarnSystemException(YarnRemoteException e) {
+	public YarnSystemException(YarnRuntimeException e) {
 		super(e.getMessage(), e);
 	}
 
@@ -55,9 +57,12 @@ public YarnSystemException(YarnRemoteException e) {
 	 *
 	 * @param e the {@link RemoteException}
 	 */
-	public YarnSystemException(RemoteException e) {
+	public YarnSystemException(IOException e) {
 		super(e.getMessage(), e);
 	}
+//	public YarnSystemException(RemoteException e) {
+//		super(e.getMessage(), e);
+//	}
 
 	/**
 	 * Constructs a general YarnSystemException.

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/am/AbstractAppmaster.java

@@ -349,7 +349,7 @@ protected RegisterApplicationMasterResponse registerAppmaster() {
 				" trackUrl=" + trackUrl + " rpcHost=" + rpcHost + " rpcPort=" + rpcPort);
 
 		RegisterApplicationMasterResponse response =
-				rmTemplate.registerApplicationMaster(applicationAttemptId, rpcHost, rpcPort, trackUrl);
+				rmTemplate.registerApplicationMaster(rpcHost, rpcPort, trackUrl);
 		applicationRegistered = true;
 		return response;
 	}
@@ -366,22 +366,26 @@ protected FinishApplicationMasterResponse finishAppmaster() {
 		boolean clean = getResourceLocalizer().clean();
 		log.info("Status of resource localizer clean operation is " + clean);
 
+		// starting from 2.1.x applicationAttemptId is part of the token and
+		// doesn't exist in finish request. We still keep it around as per
+		// old concept.
 		Assert.notNull(applicationAttemptId, "applicationAttemptId must be set");
 		if(!applicationRegistered) {
 			log.warn("Not sending finish request because we're not registered");
 			return null;
 		}
-		if(log.isDebugEnabled()) {
-			log.debug("Sending finish request to resource manager: appAttemptId=" +
-					applicationAttemptId + " status=" + FinalApplicationStatus.SUCCEEDED);
-		}
-		FinishApplicationMasterRequest finishReq = Records.newRecord(FinishApplicationMasterRequest.class);
-		finishReq.setAppAttemptId(applicationAttemptId);
 
+		FinishApplicationMasterRequest finishReq = Records.newRecord(FinishApplicationMasterRequest.class);
 		// assume succeed if not set
 		FinalApplicationStatus status = finalApplicationStatus != null ?
 				finalApplicationStatus : FinalApplicationStatus.SUCCEEDED;
-		finishReq.setFinishApplicationStatus(status);
+
+		if(log.isDebugEnabled()) {
+			log.debug("Sending finish request to resource manager. Current applicationAttemptId=" +
+					applicationAttemptId + " with status=" + status);
+		}
+
+		finishReq.setFinalApplicationStatus(status);
 		return rmTemplate.finish(finishReq);
 	}
 

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/am/AbstractProcessingAppmaster.java

@@ -82,7 +82,7 @@ public void completed(List<ContainerStatus> completedContainers) {
 	}
 
 	@Override
-	public ContainerLaunchContext preLaunch(ContainerLaunchContext context) {
+	public ContainerLaunchContext preLaunch(Container container, ContainerLaunchContext context) {
 		return context;
 	}
 

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/am/AppmasterCmOperations.java

@@ -15,9 +15,9 @@
  */
 package org.springframework.yarn.am;
 
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersResponse;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
 
 /**
@@ -36,14 +36,14 @@ public interface AppmasterCmOperations {
 	 * @param request the request
 	 * @return the start container response
 	 */
-	StartContainerResponse startContainer(StartContainerRequest request);
+	StartContainersResponse startContainers(StartContainersRequest request);
 
 	/**
 	 * Stop container.
 	 *
 	 * @return the stop container response
 	 */
-	StopContainerResponse stopContainer();
+	StopContainersResponse stopContainers();
 
 	/**
 	 * Gets the container status.

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/am/AppmasterCmTemplate.java

@@ -17,78 +17,87 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.util.ArrayList;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.net.NetUtils;
-import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.TokenIdentifier;
-import org.apache.hadoop.yarn.api.ContainerManager;
-import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StartContainerResponse;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerRequest;
-import org.apache.hadoop.yarn.api.protocolrecords.StopContainerResponse;
+import org.apache.hadoop.yarn.api.ContainerManagementProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.StopContainersResponse;
 import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.ContainerToken;
-import org.apache.hadoop.yarn.api.records.DelegationToken;
-import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
-import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
+import org.apache.hadoop.yarn.api.records.Token;
+import org.apache.hadoop.yarn.client.api.NMTokenCache;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.security.NMTokenIdentifier;
+import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hadoop.yarn.util.Records;
 import org.springframework.yarn.rpc.YarnRpcAccessor;
 import org.springframework.yarn.rpc.YarnRpcCallback;
 
 /**
  * Template implementation for {@link AppmasterCmOperations} wrapping
- * communication using {@link ContainerManager}. Methods for this
+ * communication using {@link ContainerManagementProtocol}. Methods for this
  * template wraps possible exceptions into Spring Dao exception hierarchy.
  *
  * @author Janne Valkealahti
  *
  */
-public class AppmasterCmTemplate extends YarnRpcAccessor<ContainerManager> implements AppmasterCmOperations {
+public class AppmasterCmTemplate extends YarnRpcAccessor<ContainerManagementProtocol> implements AppmasterCmOperations {
 
 	/** Container we're working for */
 	private final Container container;
 
+	/**
+	 * Instantiates a new AppmasterCmTemplate.
+	 *
+	 * @param config the hadoop configation
+	 * @param container the {@link Container}
+	 */
 	public AppmasterCmTemplate(Configuration config, Container container) {
-		super(ContainerManager.class, config);
+		super(ContainerManagementProtocol.class, config);
 		this.container = container;
 	}
 
 	@Override
-	public StartContainerResponse startContainer(final StartContainerRequest request) {
-		return execute(new YarnRpcCallback<StartContainerResponse, ContainerManager>() {
+	public StartContainersResponse startContainers(final StartContainersRequest request) {
+		return execute(new YarnRpcCallback<StartContainersResponse, ContainerManagementProtocol>() {
 			@Override
-			public StartContainerResponse doInYarn(ContainerManager proxy) throws YarnRemoteException {
-				return proxy.startContainer(request);
+			public StartContainersResponse doInYarn(ContainerManagementProtocol proxy) throws YarnException, IOException {
+				return proxy.startContainers(request);
 			}
 		});
 	}
 
 	@Override
-	public StopContainerResponse stopContainer() {
-		return execute(new YarnRpcCallback<StopContainerResponse, ContainerManager>() {
+	public StopContainersResponse stopContainers() {
+		return execute(new YarnRpcCallback<StopContainersResponse, ContainerManagementProtocol>() {
 			@Override
-			public StopContainerResponse doInYarn(ContainerManager proxy) throws YarnRemoteException {
-				StopContainerRequest request = Records.newRecord(StopContainerRequest.class);
-				request.setContainerId(container.getId());
-				return proxy.stopContainer(request);
+			public StopContainersResponse doInYarn(ContainerManagementProtocol proxy) throws YarnException, IOException {
+				StopContainersRequest request = Records.newRecord(StopContainersRequest.class);
+				ArrayList<ContainerId> ids = new ArrayList<ContainerId>();
+				ids.add(container.getId());
+				request.setContainerIds(ids);
+				return proxy.stopContainers(request);
 			}
 		});
 	}
 
 	@Override
 	public ContainerStatus getContainerStatus() {
-		return execute(new YarnRpcCallback<ContainerStatus, ContainerManager>() {
+		return execute(new YarnRpcCallback<ContainerStatus, ContainerManagementProtocol>() {
 			@Override
-			public ContainerStatus doInYarn(ContainerManager proxy) throws YarnRemoteException {
-				GetContainerStatusRequest request = Records.newRecord(GetContainerStatusRequest.class);
-				request.setContainerId(container.getId());
-				return proxy.getContainerStatus(request).getStatus();
+			public ContainerStatus doInYarn(ContainerManagementProtocol proxy) throws YarnException, IOException {
+				GetContainerStatusesRequest request = Records.newRecord(GetContainerStatusesRequest.class);
+				ArrayList<ContainerId> ids = new ArrayList<ContainerId>();
+				ids.add(container.getId());
+				request.setContainerIds(ids);
+				return proxy.getContainerStatuses(request).getContainerStatuses().get(0);
 			}
 		});
 	}
@@ -101,45 +110,19 @@ protected InetSocketAddress getRpcAddress(Configuration config) {
 
 	@Override
 	protected UserGroupInformation getUser() {
-		UserGroupInformation user = null;
-		try {
-			user = UserGroupInformation.getCurrentUser();
-			if (UserGroupInformation.isSecurityEnabled()) {
-				ContainerToken containerToken = container.getContainerToken();
-				Token<ContainerTokenIdentifier> token = null;
-				if (containerToken instanceof DelegationToken) {
-					token = convertFromProtoFormat((DelegationToken) container.getContainerToken(),
-							getRpcAddress(getConfiguration()));
-				}
-				// remote user needs to be a container id
-				user = UserGroupInformation.createRemoteUser(container.getId().toString());
-				user.addToken(token);
-			}
-		} catch (IOException e) {
-		}
-		return user;
-	}
+		InetSocketAddress rpcAddress = getRpcAddress(getConfiguration());
 
-	/**
-	 * Convert token identifier from a proto format.
-	 * <p>
-	 * This function is a copy for way it was pre hadoop-2.0.3. Helps
-	 * to work with api changes.
-	 *
-	 * @param <T> the generic type
-	 * @param protoToken the proto token
-	 * @param serviceAddr the service addr
-	 * @return the token identifier
-	 */
-	private static <T extends TokenIdentifier> Token<T> convertFromProtoFormat(DelegationToken protoToken,
-			InetSocketAddress serviceAddr) {
-		// TODO: remove this method when api's are compatible
-		Token<T> token = new Token<T>(protoToken.getIdentifier().array(), protoToken.getPassword().array(),
-				new Text(protoToken.getKind()), new Text(protoToken.getService()));
-		if (serviceAddr != null) {
-			SecurityUtil.setTokenService(token, serviceAddr);
-		}
-		return token;
+		// TODO: at some point remove static cache
+		Token token = NMTokenCache.getNMToken(container.getNodeId().toString());
+
+		// this is what node manager requires for auth
+		UserGroupInformation user =
+				UserGroupInformation.createRemoteUser(container.getId().getApplicationAttemptId().toString());
+		org.apache.hadoop.security.token.Token<NMTokenIdentifier> nmToken =
+				ConverterUtils.convertFromYarn(token, rpcAddress);
+		user.addToken(nmToken);
+
+		return user;
 	}
 
 }

spring-yarn/spring-yarn-core/src/main/java/org/springframework/yarn/am/AppmasterRmOperations.java

@@ -20,29 +20,26 @@
 import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.FinishApplicationMasterResponse;
 import org.apache.hadoop.yarn.api.protocolrecords.RegisterApplicationMasterResponse;
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 
 /**
  * Interface for appmaster to resource manager communication.
  *
  * @author Janne Valkealahti
  * @see org.springframework.yarn.am.AppmasterRmTemplate
- * @see org.apache.hadoop.yarn.api.AMRMProtocol
+ * @see org.apache.hadoop.yarn.api.ApplicationMasterProtocol
  *
  */
 public interface AppmasterRmOperations {
 
 	/**
 	 * Register application master.
 	 *
-	 * @param appAttemptId the app attempt id
 	 * @param host the host
 	 * @param rpcPort the rpc port
 	 * @param trackUrl the track url
-	 * @return the register application master response
+	 * @return the {@link RegisterApplicationMasterResponse}
 	 */
-	RegisterApplicationMasterResponse registerApplicationMaster(ApplicationAttemptId appAttemptId,
-			String host, Integer rpcPort, String trackUrl);
+	RegisterApplicationMasterResponse registerApplicationMaster(String host, Integer rpcPort, String trackUrl);
 
 	/**
 	 * Allocate container.
@@ -55,8 +52,8 @@ RegisterApplicationMasterResponse registerApplicationMaster(ApplicationAttemptId
 	/**
 	 * Finish the application master.
 	 *
-	 * @param request the request
-	 * @return the finish application master response
+	 * @param request the {@link FinishApplicationMasterRequest}
+	 * @return the {@link FinishApplicationMasterResponse}
 	 */
 	FinishApplicationMasterResponse finish(FinishApplicationMasterRequest request);