AMRMProxyService extends CompositeService implements
    ApplicationMasterProtocol { @Override public RegisterApplicationMasterResponse registerApplicationMaster( RegisterApplicationMasterRequest request) throws YarnException, IOException { long startTime = clock.getTime(); try { RequestInterceptorChainWrapper pipeline = authorizeAndGetInterceptorChain(); LOG.info("Registering application master." + " Host:" + request.getHost() + " Port:" + request.getRpcPort() + " Tracking Url:" + request .getTrackingUrl() + " for application " + pipeline .getApplicationAttemptId()); RegisterApplicationMasterResponse response = pipeline.getRootInterceptor().registerApplicationMaster(request); long endTime = clock.getTime(); this.metrics.succeededRegisterAMRequests(endTime - startTime); LOG.info("RegisterAM processing finished in {} ms for application {}", endTime - startTime, pipeline.getApplicationAttemptId()); return response; } catch (Throwable t) { this.metrics.incrFailedRegisterAMRequests(); throw t; } } AMRMProxyService(Context nmContext, AsyncDispatcher dispatcher); void recover(); @Override RegisterApplicationMasterResponse registerApplicationMaster(
      RegisterApplicationMasterRequest request); @Override FinishApplicationMasterResponse finishApplicationMaster(
      FinishApplicationMasterRequest request); @Override AllocateResponse allocate(AllocateRequest request); void processApplicationStartRequest(StartContainerRequest request); @Private InetSocketAddress getBindAddress(); @Private AMRMProxyTokenSecretManager getSecretManager();  }
@Test public void testRegisterOneApplicationMaster() throws Exception { int testAppId = 1; RegisterApplicationMasterResponse response1 = registerApplicationMaster(testAppId); Assert.assertNotNull(response1); Assert.assertEquals(Integer.toString(testAppId), response1.getQueue()); }
@Test public void testRegisterMulitpleApplicationMasters() throws Exception { for (int testAppId = 0; testAppId < 3; testAppId++) { RegisterApplicationMasterResponse response = registerApplicationMaster(testAppId); Assert.assertNotNull(response); Assert .assertEquals(Integer.toString(testAppId), response.getQueue()); } }
@Test public void testFinishMulitpleApplicationMastersInParallel() throws Exception { int numberOfRequests = 5; ArrayList<String> testContexts = new ArrayList<String>(); LOG.info("Creating " + numberOfRequests + " contexts for testing"); for (int i = 0; i < numberOfRequests; i++) { testContexts.add("test-endpoint-" + Integer.toString(i)); LOG.info("Created test context: " + testContexts.get(i)); RegisterApplicationMasterResponse registerResponse = registerApplicationMaster(i); Assert.assertNotNull(registerResponse); Assert .assertEquals(Integer.toString(i), registerResponse.getQueue()); } finishApplicationMastersInParallel(testContexts); }
@Test public void testAppRecoveryFailure() throws YarnException, Exception { Configuration conf = createConfiguration(); conf.set(YarnConfiguration.AMRM_PROXY_INTERCEPTOR_CLASS_PIPELINE, BadRequestInterceptorAcrossRestart.class.getName()); mockRM = new MockResourceManagerFacade(new YarnConfiguration(conf), 0); createAndStartAMRMProxyService(conf); registerApplicationMaster(1); RecoveredAMRMProxyState state = getNMContext().getNMStateStore().loadAMRMProxyState(); Assert.assertEquals(1, state.getAppContexts().size()); createAndStartAMRMProxyService(conf); state = getNMContext().getNMStateStore().loadAMRMProxyState(); Assert.assertEquals(0, state.getAppContexts().size()); }