java类org.apache.hadoop.yarn.api.records.ContainerLaunchContext的实例源码

@Test
public void testContainerKill() throws IOException {
  String appSubmitter = "nobody";
  String cmd = String.valueOf(
      PrivilegedOperation.RunAsUserCommand.SIGNAL_CONTAINER.getValue());
  ContainerExecutor.Signal signal = ContainerExecutor.Signal.QUIT;
  String sigVal = String.valueOf(signal.getValue());

  Container container = mock(Container.class);
  ContainerId cId = mock(ContainerId.class);
  ContainerLaunchContext context = mock(ContainerLaunchContext.class);

  when(container.getContainerId()).thenReturn(cId);
  when(container.getLaunchContext()).thenReturn(context);

  mockExec.signalContainer(new ContainerSignalContext.Builder()
      .setContainer(container)
      .setUser(appSubmitter)
      .setPid("1000")
      .setSignal(signal)
      .build());
  assertEquals(Arrays.asList(YarnConfiguration.DEFAULT_NM_NONSECURE_MODE_LOCAL_USER,
      appSubmitter, cmd, "1000", sigVal),
      readMockParams());
}

public boolean run() throws Exception {
  YarnClientApplication app = createApplication();
  ApplicationId appId = app.getNewApplicationResponse().getApplicationId();

  // Copy the application jar to the filesystem
  FileSystem fs = FileSystem.get(conf);
  String appIdStr = appId.toString();
  Path dstJarPath = Utils.copyLocalFileToDfs(fs, appIdStr, new Path(tfJar), Constants.TF_JAR_NAME);
  Path dstLibPath = Utils.copyLocalFileToDfs(fs, appIdStr, new Path(tfLib),
      Constants.TF_LIB_NAME);
  Map<String, Path> files = new HashMap<>();
  files.put(Constants.TF_JAR_NAME, dstJarPath);
  Map<String, LocalResource> localResources = Utils.makeLocalResources(fs, files);
  Map<String, String> javaEnv = Utils.setJavaEnv(conf);
  String command = makeAppMasterCommand(dstLibPath.toString(), dstJarPath.toString());
  LOG.info("Make ApplicationMaster command: " + command);
  ContainerLaunchContext launchContext = ContainerLaunchContext.newInstance(
      localResources, javaEnv, Lists.newArrayList(command), null, null, null);
  Resource resource = Resource.newInstance(amMemory, amVCores);
  submitApplication(app, appName, launchContext, resource, amQueue);
  return awaitApplication(appId);
}

public static ApplicationSubmissionContext newApplicationSubmissionContext(
    ApplicationId applicationId, String applicationName, String queue,
    Priority priority, ContainerLaunchContext amContainer,
    boolean isUnmanagedAM, boolean cancelTokensWhenComplete,
    int maxAppAttempts, Resource resource, String applicationType) {
  ApplicationSubmissionContext context =
      recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
  context.setApplicationId(applicationId);
  context.setApplicationName(applicationName);
  context.setQueue(queue);
  context.setPriority(priority);
  context.setAMContainerSpec(amContainer);
  context.setUnmanagedAM(isUnmanagedAM);
  context.setCancelTokensWhenComplete(cancelTokensWhenComplete);
  context.setMaxAppAttempts(maxAppAttempts);
  context.setResource(resource);
  context.setApplicationType(applicationType);
  return context;
}

private void startContainer(final YarnRPC rpc,
    org.apache.hadoop.yarn.api.records.Token nmToken,
    org.apache.hadoop.yarn.api.records.Token containerToken,
    NodeId nodeId, String user) throws Exception {

  ContainerLaunchContext context =
      Records.newRecord(ContainerLaunchContext.class);
  StartContainerRequest scRequest =
      StartContainerRequest.newInstance(context,containerToken);
  List<StartContainerRequest> list = new ArrayList<StartContainerRequest>();
  list.add(scRequest);
  StartContainersRequest allRequests =
      StartContainersRequest.newInstance(list);
  ContainerManagementProtocol proxy = null;
  try {
    proxy = getContainerManagementProtocolProxy(rpc, nmToken, nodeId, user);
    StartContainersResponse response = proxy.startContainers(allRequests);
    for(SerializedException ex : response.getFailedRequests().values()){
      parseAndThrowException(ex.deSerialize());
    }
  } finally {
    if (proxy != null) {
      rpc.stopProxy(proxy, conf);
    }
  }
}

private Credentials parseCredentials(ContainerLaunchContext launchContext)
    throws IOException {
  Credentials credentials = new Credentials();
  // //////////// Parse credentials
  ByteBuffer tokens = launchContext.getTokens();

  if (tokens != null) {
    DataInputByteBuffer buf = new DataInputByteBuffer();
    tokens.rewind();
    buf.reset(tokens);
    credentials.readTokenStorageStream(buf);
    if (LOG.isDebugEnabled()) {
      for (Token<? extends TokenIdentifier> tk : credentials.getAllTokens()) {
        LOG.debug(tk.getService() + " = " + tk.toString());
      }
    }
  }
  // //////////// End of parsing credentials
  return credentials;
}

public ContainerImpl(Configuration conf, Dispatcher dispatcher,
    NMStateStoreService stateStore, ContainerLaunchContext launchContext,
    Credentials creds, NodeManagerMetrics metrics,
    ContainerTokenIdentifier containerTokenIdentifier) {
  this.daemonConf = conf;
  this.dispatcher = dispatcher;
  this.stateStore = stateStore;
  this.launchContext = launchContext;
  this.containerTokenIdentifier = containerTokenIdentifier;
  this.containerId = containerTokenIdentifier.getContainerID();
  this.resource = containerTokenIdentifier.getResource();
  this.diagnostics = new StringBuilder();
  this.credentials = creds;
  this.metrics = metrics;
  user = containerTokenIdentifier.getApplicationSubmitter();
  ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  this.readLock = readWriteLock.readLock();
  this.writeLock = readWriteLock.writeLock();

  stateMachine = stateMachineFactory.make(this);
}

public MockContainer(ApplicationAttemptId appAttemptId,
    Dispatcher dispatcher, Configuration conf, String user,
    ApplicationId appId, int uniqId) throws IOException{

  this.user = user;
  this.recordFactory = RecordFactoryProvider.getRecordFactory(conf);
  this.id = BuilderUtils.newContainerId(recordFactory, appId, appAttemptId,
      uniqId);
  this.launchContext = recordFactory
      .newRecordInstance(ContainerLaunchContext.class);
  long currentTime = System.currentTimeMillis();
  this.containerTokenIdentifier =
      BuilderUtils.newContainerTokenIdentifier(BuilderUtils
        .newContainerToken(id, "127.0.0.1", 1234, user,
          BuilderUtils.newResource(1024, 1), currentTime + 10000, 123,
          "password".getBytes(), currentTime));
  this.state = ContainerState.NEW;
}

@SuppressWarnings("rawtypes")
@Test (timeout = 10000)
public void testCallFailureWithNullLocalizedResources() {
  Container container = mock(Container.class);
  when(container.getContainerId()).thenReturn(ContainerId.newContainerId(
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(
          System.currentTimeMillis(), 1), 1), 1));
  ContainerLaunchContext clc = mock(ContainerLaunchContext.class);
  when(clc.getCommands()).thenReturn(Collections.<String>emptyList());
  when(container.getLaunchContext()).thenReturn(clc);
  when(container.getLocalizedResources()).thenReturn(null);
  Dispatcher dispatcher = mock(Dispatcher.class);
  EventHandler eventHandler = new EventHandler() {
    public void handle(Event event) {
      Assert.assertTrue(event instanceof ContainerExitEvent);
      ContainerExitEvent exitEvent = (ContainerExitEvent) event;
      Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
          exitEvent.getType());
    }
  };
  when(dispatcher.getEventHandler()).thenReturn(eventHandler);
  ContainerLaunch launch = new ContainerLaunch(context, new Configuration(),
      dispatcher, exec, null, container, dirsHandler, containerManager);
  launch.call();
}

private StartContainersResponse startContainer(Context context,
    final ContainerManagerImpl cm, ContainerId cid,
    ContainerLaunchContext clc, LogAggregationContext logAggregationContext)
        throws Exception {
  UserGroupInformation user = UserGroupInformation.createRemoteUser(
      cid.getApplicationAttemptId().toString());
  StartContainerRequest scReq = StartContainerRequest.newInstance(
      clc, TestContainerManager.createContainerToken(cid, 0,
          context.getNodeId(), user.getShortUserName(),
          context.getContainerTokenSecretManager(), logAggregationContext));
  final List<StartContainerRequest> scReqList =
      new ArrayList<StartContainerRequest>();
  scReqList.add(scReq);
  NMTokenIdentifier nmToken = new NMTokenIdentifier(
      cid.getApplicationAttemptId(), context.getNodeId(),
      user.getShortUserName(),
      context.getNMTokenSecretManager().getCurrentKey().getKeyId());
  user.addTokenIdentifier(nmToken);
  return user.doAs(new PrivilegedExceptionAction<StartContainersResponse>() {
    @Override
    public StartContainersResponse run() throws Exception {
      return cm.startContainers(
          StartContainersRequest.newInstance(scReqList));
    }
  });
}

private ContainerLaunchContext createAMContainerLaunchContext(
    ApplicationSubmissionContext applicationMasterContext,
    ContainerId containerID) throws IOException {

  // Construct the actual Container
  ContainerLaunchContext container = 
      applicationMasterContext.getAMContainerSpec();
  LOG.info("Command to launch container "
      + containerID
      + " : "
      + StringUtils.arrayToString(container.getCommands().toArray(
          new String[0])));

  // Finalize the container
  setupTokens(container, containerID);

  return container;
}

@Test (timeout = 30000)
public void testAppRecoverPath() throws IOException {
  LOG.info("--- START: testAppRecoverPath ---");
  ApplicationSubmissionContext sub =
      Records.newRecord(ApplicationSubmissionContext.class);
  ContainerLaunchContext clc =
      Records.newRecord(ContainerLaunchContext.class);
  Credentials credentials = new Credentials();
  DataOutputBuffer dob = new DataOutputBuffer();
  credentials.writeTokenStorageToStream(dob);
  ByteBuffer securityTokens =
      ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
  clc.setTokens(securityTokens);
  sub.setAMContainerSpec(clc);
  testCreateAppSubmittedRecovery(sub);
}

@SuppressWarnings("deprecation")
private SubmitApplicationRequest mockSubmitAppRequest(ApplicationId appId,
      String name, String queue, Set<String> tags, boolean unmanaged) {

  ContainerLaunchContext amContainerSpec = mock(ContainerLaunchContext.class);

  Resource resource = Resources.createResource(
      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);

  ApplicationSubmissionContext submissionContext =
      recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
  submissionContext.setAMContainerSpec(amContainerSpec);
  submissionContext.setApplicationName(name);
  submissionContext.setQueue(queue);
  submissionContext.setApplicationId(appId);
  submissionContext.setResource(resource);
  submissionContext.setApplicationType(appType);
  submissionContext.setApplicationTags(tags);
  submissionContext.setUnmanagedAM(unmanaged);

  SubmitApplicationRequest submitRequest =
      recordFactory.newRecordInstance(SubmitApplicationRequest.class);
  submitRequest.setApplicationSubmissionContext(submissionContext);
  return submitRequest;
}

public void startContainerAsync(
    Container container, ContainerLaunchContext containerLaunchContext) {
  if (containers.putIfAbsent(container.getId(),
      new StatefulContainer(this, container.getId())) != null) {
    callbackHandler.onStartContainerError(container.getId(),
        RPCUtil.getRemoteException("Container " + container.getId() +
            " is already started or scheduled to start"));
  }
  try {
    events.put(new StartContainerEvent(container, containerLaunchContext));
  } catch (InterruptedException e) {
    LOG.warn("Exception when scheduling the event of starting Container " +
        container.getId());
    callbackHandler.onStartContainerError(container.getId(), e);
  }
}

@Test(timeout = 15000)
public void testSubmitApplicationOnHA() throws Exception {
  ApplicationSubmissionContext appContext =
      Records.newRecord(ApplicationSubmissionContext.class);
  appContext.setApplicationId(cluster.createFakeAppId());
  ContainerLaunchContext amContainer =
      Records.newRecord(ContainerLaunchContext.class);
  appContext.setAMContainerSpec(amContainer);
  Resource capability = Records.newRecord(Resource.class);
  capability.setMemory(10);
  capability.setVirtualCores(1);
  capability.setGpuCores(1);
  appContext.setResource(capability);
  ApplicationId appId = client.submitApplication(appContext);
  Assert.assertTrue(getActiveRM().getRMContext().getRMApps()
      .containsKey(appId));
}

@Override
protected ContainerLauncher createContainerLauncher(AppContext context) {
  return new MockContainerLauncher() {
    @Override
    public void handle(ContainerLauncherEvent event) {
      if (event.getType() == EventType.CONTAINER_REMOTE_LAUNCH) {
        ContainerRemoteLaunchEvent launchEvent = (ContainerRemoteLaunchEvent) event;
        ContainerLaunchContext launchContext =
            launchEvent.getContainerLaunchContext();
        String cmdString = launchContext.getCommands().toString();
        LOG.info("launchContext " + cmdString);
        myCommandLine = cmdString;
        cmdEnvironment = launchContext.getEnvironment();
      }
      super.handle(event);
    }
  };
}

@Test(timeout=20000)
public void testAMProfiler() throws Exception {
  JobConf jobConf = new JobConf();

  jobConf.setBoolean(MRJobConfig.MR_AM_PROFILE, true);

  YARNRunner yarnRunner = new YARNRunner(jobConf);

  ApplicationSubmissionContext submissionContext =
      buildSubmitContext(yarnRunner, jobConf);

  ContainerLaunchContext containerSpec = submissionContext.getAMContainerSpec();
  List<String> commands = containerSpec.getCommands();

  for(String command : commands) {
    if (command != null) {
      if (command.contains(PROFILE_PARAMS)) {
        return;
      }
    }
  }
  throw new IllegalStateException("Profiler opts not found!");
}

private synchronized void launchDummyTask(Container container){
    ContainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class);
    String new_command = "./launcher.py";
    String cmd = new_command + " 1>"
        + ApplicationConstants.LOG_DIR_EXPANSION_VAR + "/stdout"
        + " 2>" + ApplicationConstants.LOG_DIR_EXPANSION_VAR
        + "/stderr";
    ctx.setCommands(Collections.singletonList(cmd));
    ctx.setTokens(setupTokens());
    ctx.setLocalResources(this.workerResources);
    synchronized (this){
        this.nmClient.startContainerAsync(container, ctx);
    }
}

@Test(timeout=20000)
public void testAMProfiler() throws Exception {
  JobConf jobConf = new JobConf();

  jobConf.setBoolean(MRJobConfig.MR_AM_PROFILE, true);

  YARNRunner yarnRunner = new YARNRunner(jobConf);

  ApplicationSubmissionContext submissionContext =
      buildSubmitContext(yarnRunner, jobConf);

  ContainerLaunchContext containerSpec = submissionContext.getAMContainerSpec();
  List<String> commands = containerSpec.getCommands();

  for(String command : commands) {
    if (command != null) {
      if (command.contains(PROFILE_PARAMS)) {
        return;
      }
    }
  }
  throw new IllegalStateException("Profiler opts not found!");
}

public static ApplicationSubmissionContext newApplicationSubmissionContext(
    ApplicationId applicationId, String applicationName, String queue,
    Priority priority, ContainerLaunchContext amContainer,
    boolean isUnmanagedAM, boolean cancelTokensWhenComplete,
    int maxAppAttempts, Resource resource, String applicationType) {
  ApplicationSubmissionContext context =
      recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
  context.setApplicationId(applicationId);
  context.setApplicationName(applicationName);
  context.setQueue(queue);
  context.setPriority(priority);
  context.setAMContainerSpec(amContainer);
  context.setUnmanagedAM(isUnmanagedAM);
  context.setCancelTokensWhenComplete(cancelTokensWhenComplete);
  context.setMaxAppAttempts(maxAppAttempts);
  context.setResource(resource);
  context.setApplicationType(applicationType);
  return context;
}

private int runAndBlock(ContainerId cId, String ... cmd) throws IOException {
  String appId = "APP_"+getNextId();
  Container container = mock(Container.class);
  ContainerLaunchContext context = mock(ContainerLaunchContext.class);
  HashMap<String, String> env = new HashMap<String,String>();

  when(container.getContainerId()).thenReturn(cId);
  when(container.getLaunchContext()).thenReturn(context);

  when(context.getEnvironment()).thenReturn(env);

  String script = writeScriptFile(cmd);

  Path scriptPath = new Path(script);
  Path tokensPath = new Path("/dev/null");
  Path workDir = new Path(workSpace.getAbsolutePath());
  Path pidFile = new Path(workDir, "pid.txt");

  exec.activateContainer(cId, pidFile);
  return exec.launchContainer(container, scriptPath, tokensPath,
      appSubmitter, appId, workDir, dirsHandler.getLocalDirs(),
      dirsHandler.getLogDirs());
}

public ContainerImpl(Configuration conf, Dispatcher dispatcher,
    NMStateStoreService stateStore, ContainerLaunchContext launchContext,
    Credentials creds, NodeManagerMetrics metrics,
    ContainerTokenIdentifier containerTokenIdentifier,
    RecoveredContainerStatus recoveredStatus, int exitCode,
    String diagnostics, boolean wasKilled, Resource recoveredCapability) {
  this(conf, dispatcher, stateStore, launchContext, creds, metrics,
      containerTokenIdentifier);
  this.recoveredStatus = recoveredStatus;
  this.exitCode = exitCode;
  this.recoveredAsKilled = wasKilled;
  this.diagnostics.append(diagnostics);
  if (recoveredCapability != null
      && !this.resource.equals(recoveredCapability)) {
    // resource capability had been updated before NM was down
    this.resource = Resource.newInstance(recoveredCapability.getMemory(),
        recoveredCapability.getVirtualCores());
  }
}

@SuppressWarnings("rawtypes")
@Test (timeout = 10000)
public void testCallFailureWithNullLocalizedResources() {
  Container container = mock(Container.class);
  when(container.getContainerId()).thenReturn(ContainerId.newContainerId(
      ApplicationAttemptId.newInstance(ApplicationId.newInstance(
          System.currentTimeMillis(), 1), 1), 1));
  ContainerLaunchContext clc = mock(ContainerLaunchContext.class);
  when(clc.getCommands()).thenReturn(Collections.<String>emptyList());
  when(container.getLaunchContext()).thenReturn(clc);
  when(container.getLocalizedResources()).thenReturn(null);
  Dispatcher dispatcher = mock(Dispatcher.class);
  EventHandler eventHandler = new EventHandler() {
    public void handle(Event event) {
      Assert.assertTrue(event instanceof ContainerExitEvent);
      ContainerExitEvent exitEvent = (ContainerExitEvent) event;
      Assert.assertEquals(ContainerEventType.CONTAINER_EXITED_WITH_FAILURE,
          exitEvent.getType());
    }
  };
  when(dispatcher.getEventHandler()).thenReturn(eventHandler);
  ContainerLaunch launch = new ContainerLaunch(context, new Configuration(),
      dispatcher, exec, null, container, dirsHandler, containerManager);
  launch.call();
}

private StartContainersResponse startContainer(Context context,
    final ContainerManagerImpl cm, ContainerId cid,
    ContainerLaunchContext clc, LogAggregationContext logAggregationContext)
        throws Exception {
  UserGroupInformation user = UserGroupInformation.createRemoteUser(
      cid.getApplicationAttemptId().toString());
  StartContainerRequest scReq = StartContainerRequest.newInstance(
      clc, TestContainerManager.createContainerToken(cid, 0,
          context.getNodeId(), user.getShortUserName(),
          context.getContainerTokenSecretManager(), logAggregationContext));
  final List<StartContainerRequest> scReqList =
      new ArrayList<StartContainerRequest>();
  scReqList.add(scReq);
  NMTokenIdentifier nmToken = new NMTokenIdentifier(
      cid.getApplicationAttemptId(), context.getNodeId(),
      user.getShortUserName(),
      context.getNMTokenSecretManager().getCurrentKey().getKeyId());
  user.addTokenIdentifier(nmToken);
  return user.doAs(new PrivilegedExceptionAction<StartContainersResponse>() {
    @Override
    public StartContainersResponse run() throws Exception {
      return cm.startContainers(
          StartContainersRequest.newInstance(scReqList));
    }
  });
}

private ContainerLaunchContext createAMContainerLaunchContext(
    ApplicationSubmissionContext applicationMasterContext,
    ContainerId containerID) throws IOException {

  // Construct the actual Container
  ContainerLaunchContext container =
      applicationMasterContext.getAMContainerSpec();
  LOG.info("Command to launch container "
      + containerID
      + " : "
      + StringUtils.arrayToString(container.getCommands().toArray(
          new String[0])));

  // Finalize the container
  setupTokens(container, containerID);

  return container;
}

@Test (timeout = 30000)
public void testAppRecoverPath() throws IOException {
  LOG.info("--- START: testAppRecoverPath ---");
  ApplicationSubmissionContext sub =
      Records.newRecord(ApplicationSubmissionContext.class);
  ContainerLaunchContext clc =
      Records.newRecord(ContainerLaunchContext.class);
  Credentials credentials = new Credentials();
  DataOutputBuffer dob = new DataOutputBuffer();
  credentials.writeTokenStorageToStream(dob);
  ByteBuffer securityTokens =
      ByteBuffer.wrap(dob.getData(), 0, dob.getLength());
  clc.setTokens(securityTokens);
  sub.setAMContainerSpec(clc);
  testCreateAppSubmittedRecovery(sub);
}

@Test(timeout = 15000)
public void testSubmitApplicationOnHA() throws Exception {
  ApplicationSubmissionContext appContext =
      Records.newRecord(ApplicationSubmissionContext.class);
  appContext.setApplicationId(cluster.createFakeAppId());
  ContainerLaunchContext amContainer =
      Records.newRecord(ContainerLaunchContext.class);
  appContext.setAMContainerSpec(amContainer);
  Resource capability = Records.newRecord(Resource.class);
  capability.setMemory(10);
  capability.setVirtualCores(1);
  appContext.setResource(capability);
  ApplicationId appId = client.submitApplication(appContext);
  Assert.assertTrue(getActiveRM().getRMContext().getRMApps()
      .containsKey(appId));
}

public void startContainerAsync(
    Container container, ContainerLaunchContext containerLaunchContext) {
  if (containers.putIfAbsent(container.getId(),
      new StatefulContainer(this, container.getId())) != null) {
    callbackHandler.onStartContainerError(container.getId(),
        RPCUtil.getRemoteException("Container " + container.getId() +
            " is already started or scheduled to start"));
  }
  try {
    events.put(new StartContainerEvent(container, containerLaunchContext));
  } catch (InterruptedException e) {
    LOG.warn("Exception when scheduling the event of starting Container " +
        container.getId());
    callbackHandler.onStartContainerError(container.getId(), e);
  }
}

@SuppressWarnings("deprecation")
private SubmitApplicationRequest mockSubmitAppRequest(ApplicationId appId,
      String name, String queue, Set<String> tags, boolean unmanaged) {

  ContainerLaunchContext amContainerSpec = mock(ContainerLaunchContext.class);

  Resource resource = Resources.createResource(
      YarnConfiguration.DEFAULT_RM_SCHEDULER_MINIMUM_ALLOCATION_MB);

  ApplicationSubmissionContext submissionContext =
      recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
  submissionContext.setAMContainerSpec(amContainerSpec);
  submissionContext.setApplicationName(name);
  submissionContext.setQueue(queue);
  submissionContext.setApplicationId(appId);
  submissionContext.setResource(resource);
  submissionContext.setApplicationType(appType);
  submissionContext.setApplicationTags(tags);
  submissionContext.setUnmanagedAM(unmanaged);

  SubmitApplicationRequest submitRequest =
      recordFactory.newRecordInstance(SubmitApplicationRequest.class);
  submitRequest.setApplicationSubmissionContext(submissionContext);
  return submitRequest;
}

@Test(timeout = 15000)
public void testSubmitApplicationOnHA() throws Exception {
  ApplicationSubmissionContext appContext =
      Records.newRecord(ApplicationSubmissionContext.class);
  appContext.setApplicationId(cluster.createFakeAppId());
  ContainerLaunchContext amContainer =
      Records.newRecord(ContainerLaunchContext.class);
  appContext.setAMContainerSpec(amContainer);
  Resource capability = Records.newRecord(Resource.class);
  capability.setMemory(10);
  capability.setVirtualCores(1);
  appContext.setResource(capability);
  ApplicationId appId = client.submitApplication(appContext);
  Assert.assertTrue(getActiveRM().getRMContext().getRMApps()
      .containsKey(appId));
}

private StartContainersResponse startContainer(Context context,
    final ContainerManagerImpl cm, ContainerId cid,
    ContainerLaunchContext clc, LogAggregationContext logAggregationContext)
        throws Exception {
  UserGroupInformation user = UserGroupInformation.createRemoteUser(
      cid.getApplicationAttemptId().toString());
  StartContainerRequest scReq = StartContainerRequest.newInstance(
      clc, TestContainerManager.createContainerToken(cid, 0,
          context.getNodeId(), user.getShortUserName(),
          context.getContainerTokenSecretManager(), logAggregationContext));
  final List<StartContainerRequest> scReqList =
      new ArrayList<StartContainerRequest>();
  scReqList.add(scReq);
  NMTokenIdentifier nmToken = new NMTokenIdentifier(
      cid.getApplicationAttemptId(), context.getNodeId(),
      user.getShortUserName(),
      context.getNMTokenSecretManager().getCurrentKey().getKeyId());
  user.addTokenIdentifier(nmToken);
  return user.doAs(new PrivilegedExceptionAction<StartContainersResponse>() {
    @Override
    public StartContainersResponse run() throws Exception {
      return cm.startContainers(
          StartContainersRequest.newInstance(scReqList));
    }
  });
}