MRAppMaster extends CompositeService { protected static void initAndStartAppMaster(final MRAppMaster appMaster, final JobConf conf, String jobUserName) throws IOException, InterruptedException { UserGroupInformation.setConfiguration(conf); Credentials credentials = UserGroupInformation.getCurrentUser().getCredentials(); LOG.info("Executing with tokens:"); for (Token<?> token : credentials.getAllTokens()) { LOG.info(token); } UserGroupInformation appMasterUgi = UserGroupInformation .createRemoteUser(jobUserName); appMasterUgi.addCredentials(credentials); Iterator<Token<?>> iter = credentials.getAllTokens().iterator(); while (iter.hasNext()) { Token<?> token = iter.next(); if (token.getKind().equals(AMRMTokenIdentifier.KIND_NAME)) { iter.remove(); } } conf.getCredentials().addAll(credentials); appMasterUgi.doAs(new PrivilegedExceptionAction<Object>() { @Override public Object run() throws Exception { appMaster.init(conf); appMaster.start(); if(appMaster.errorHappenedShutDown) { throw new IOException("Was asked to shut down."); } return null; } }); } MRAppMaster(ApplicationAttemptId applicationAttemptId,
      ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
      long appSubmitTime);  MRAppMaster(ApplicationAttemptId applicationAttemptId,
      ContainerId containerId, String nmHost, int nmPort, int nmHttpPort,
      Clock clock, long appSubmitTime); void cleanupStagingDir(); @VisibleForTesting void shutDownJob(); ApplicationId getAppID(); ApplicationAttemptId getAttemptID(); JobId getJobId(); OutputCommitter getCommitter(); boolean isNewApiCommitter(); int getStartCount(); AppContext getContext(); Dispatcher getDispatcher(); Map<TaskId, TaskInfo> getCompletedTaskFromPreviousRun(); List<AMInfo> getAllAMInfos(); ContainerAllocator getContainerAllocator(); ContainerLauncher getContainerLauncher(); TaskAttemptListener getTaskAttemptListener(); Boolean isLastAMRetry(); @Override void stop(); static void main(String[] args); void notifyIsLastAMRetry(boolean isLastAMRetry); ClientService getClientService(); static final int SHUTDOWN_HOOK_PRIORITY; static final String INTERMEDIATE_DATA_ENCRYPTION_ALGO; }
@Test public void testMRAppMasterForDifferentUser() throws IOException, InterruptedException { String applicationAttemptIdStr = "appattempt_1317529182569_0004_000001"; String containerIdStr = "container_1317529182569_0004_000001_1"; String userName = "TestAppMasterUser"; ApplicationAttemptId applicationAttemptId = ConverterUtils .toApplicationAttemptId(applicationAttemptIdStr); ContainerId containerId = ConverterUtils.toContainerId(containerIdStr); MRAppMasterTest appMaster = new MRAppMasterTest(applicationAttemptId, containerId, "host", -1, -1, System.currentTimeMillis()); JobConf conf = new JobConf(); conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir); MRAppMaster.initAndStartAppMaster(appMaster, conf, userName); Path userPath = new Path(stagingDir, userName); Path userStagingPath = new Path(userPath, ".staging"); assertEquals(userStagingPath.toString(), appMaster.stagingDirPath.toString()); }