ApplicationMasterService extends AbstractService implements
    ApplicationMasterProtocol { public void registerAppAttempt(ApplicationAttemptId attemptId) { AllocateResponse response = recordFactory.newRecordInstance(AllocateResponse.class); response.setResponseId(AMRMClientUtils.PRE_REGISTER_RESPONSE_ID); LOG.info("Registering app attempt : " + attemptId); responseMap.put(attemptId, new AllocateResponseLock(response)); rmContext.getNMTokenSecretManager().registerApplicationAttempt(attemptId); } ApplicationMasterService(RMContext rmContext,
      YarnScheduler scheduler);  ApplicationMasterService(String name, RMContext rmContext,
      YarnScheduler scheduler); @Private InetSocketAddress getBindAddress(); @Override RegisterApplicationMasterResponse registerApplicationMaster(
      RegisterApplicationMasterRequest request); @Override FinishApplicationMasterResponse finishApplicationMaster(
      FinishApplicationMasterRequest request); boolean hasApplicationMasterRegistered(
      ApplicationAttemptId appAttemptId); @Override AllocateResponse allocate(AllocateRequest request); void registerAppAttempt(ApplicationAttemptId attemptId); void unregisterAttempt(ApplicationAttemptId attemptId); void refreshServiceAcls(Configuration configuration, 
      PolicyProvider policyProvider); @VisibleForTesting Server getServer();  }
@Test(timeout = 300000) public void testApplicationMasterInterceptor() throws Exception { conf.set(YarnConfiguration.RM_APPLICATION_MASTER_SERVICE_PROCESSORS, TestInterceptor1.class.getName() + "," + TestInterceptor2.class.getName()); MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); int allocCount = 0; am1.addRequests(new String[] {"127.0.0.1"}, GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); allocCount++; nm1.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 1..."); sleep(1000); alloc1Response = am1.schedule(); allocCount++; } Container allocatedContainer = alloc1Response.getAllocatedContainers().get(0); ContainerTokenIdentifier tokenId = BuilderUtils.newContainerTokenIdentifier(allocatedContainer .getContainerToken()); am1.unregisterAppAttempt(); Assert.assertEquals(1, beforeRegCount.get()); Assert.assertEquals(1, afterRegCount.get()); Assert.assertEquals(allocCount * 2, beforeAllocCount.get()); Assert.assertEquals(allocCount * 2, afterAllocCount.get()); Assert.assertEquals(1, beforeFinishCount.get()); Assert.assertEquals(1, afterFinishCount.get()); rm.stop(); }
@Test(timeout = 3000000) public void testRMIdentifierOnContainerAllocation() throws Exception { MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); am1.addRequests(new String[] { "127.0.0.1" }, GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); nm1.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 1..."); sleep(1000); alloc1Response = am1.schedule(); } Container allocatedContainer = alloc1Response.getAllocatedContainers().get(0); ContainerTokenIdentifier tokenId = BuilderUtils.newContainerTokenIdentifier(allocatedContainer .getContainerToken()); Assert.assertEquals(MockRM.getClusterTimeStamp(), tokenId.getRMIdentifier()); rm.stop(); }
@Test(timeout = 3000000) public void testAllocateResponseIdOverflow() throws Exception { MockRM rm = new MockRM(conf); try { rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); Assert.assertTrue(am1.setApplicationLastResponseId(Integer.MAX_VALUE)); am1.schedule(); Assert.assertEquals(0, am1.getResponseId()); am1.schedule(); Assert.assertEquals(1, am1.getResponseId()); } finally { if (rm != null) { rm.stop(); } } }
@Test(timeout=600000) public void testInvalidContainerReleaseRequest() throws Exception { MockRM rm = new MockRM(conf); try { rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(1024); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); am1.addRequests(new String[] { "127.0.0.1" }, GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); nm1.nodeHeartbeat(true); while (alloc1Response.getAllocatedContainers().size() < 1) { LOG.info("Waiting for containers to be created for app 1..."); sleep(1000); alloc1Response = am1.schedule(); } Assert.assertTrue(alloc1Response.getAllocatedContainers().size() > 0); RMApp app2 = rm.submitApp(1024); nm1.nodeHeartbeat(true); RMAppAttempt attempt2 = app2.getCurrentAppAttempt(); MockAM am2 = rm.sendAMLaunched(attempt2.getAppAttemptId()); am2.registerAppAttempt(); ContainerId cId = alloc1Response.getAllocatedContainers().get(0).getId(); am2.addContainerToBeReleased(cId); try { am2.schedule(); fail("Exception was expected!!"); } catch (InvalidContainerReleaseException e) { StringBuilder sb = new StringBuilder("Cannot release container : "); sb.append(cId.toString()); sb.append(" not belonging to this application attempt : "); sb.append(attempt2.getAppAttemptId().toString()); Assert.assertTrue(e.getMessage().contains(sb.toString())); } } finally { if (rm != null) { rm.stop(); } } }
@Test(timeout=1200000) public void testFinishApplicationMasterBeforeRegistering() throws Exception { MockRM rm = new MockRM(conf); try { rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); MockAM am1 = MockRM.launchAM(app1, rm, nm1); FinishApplicationMasterRequest req = FinishApplicationMasterRequest.newInstance( FinalApplicationStatus.FAILED, "", ""); try { am1.unregisterAppAttempt(req, false); fail("ApplicationMasterNotRegisteredException should be thrown"); } catch (ApplicationMasterNotRegisteredException e) { Assert.assertNotNull(e); Assert.assertNotNull(e.getMessage()); Assert.assertTrue(e.getMessage().contains( "Application Master is trying to unregister before registering for:" )); } catch (Exception e) { fail("ApplicationMasterNotRegisteredException should be thrown"); } am1.registerAppAttempt(); am1.unregisterAppAttempt(req, false); rm.waitForState(am1.getApplicationAttemptId(), RMAppAttemptState.FINISHING); } finally { if (rm != null) { rm.stop(); } } }
@Test(timeout = 3000000) public void testResourceTypes() throws Exception { HashMap<YarnConfiguration, EnumSet<SchedulerResourceTypes>> driver = new HashMap<YarnConfiguration, EnumSet<SchedulerResourceTypes>>(); CapacitySchedulerConfiguration csconf = new CapacitySchedulerConfiguration(); csconf.setResourceComparator(DominantResourceCalculator.class); YarnConfiguration testCapacityDRConf = new YarnConfiguration(csconf); testCapacityDRConf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); YarnConfiguration testCapacityDefConf = new YarnConfiguration(); testCapacityDefConf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); YarnConfiguration testFairDefConf = new YarnConfiguration(); testFairDefConf.setClass(YarnConfiguration.RM_SCHEDULER, FairScheduler.class, ResourceScheduler.class); driver.put(conf, EnumSet.of(SchedulerResourceTypes.MEMORY)); driver.put(testCapacityDRConf, EnumSet.of(SchedulerResourceTypes.CPU, SchedulerResourceTypes.MEMORY)); driver.put(testCapacityDefConf, EnumSet.of(SchedulerResourceTypes.MEMORY)); driver.put(testFairDefConf, EnumSet.of(SchedulerResourceTypes.MEMORY, SchedulerResourceTypes.CPU)); for (Map.Entry<YarnConfiguration, EnumSet<SchedulerResourceTypes>> entry : driver .entrySet()) { EnumSet<SchedulerResourceTypes> expectedValue = entry.getValue(); MockRM rm = new MockRM(entry.getKey()); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); Thread.sleep(1000); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); RegisterApplicationMasterResponse resp = am1.registerAppAttempt(); EnumSet<SchedulerResourceTypes> types = resp.getSchedulerResourceTypes(); LOG.info("types = " + types.toString()); Assert.assertEquals(expectedValue, types); rm.stop(); } }
@Test(timeout=1200000) public void testAllocateAfterUnregister() throws Exception { MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); FinishApplicationMasterRequest req = FinishApplicationMasterRequest.newInstance( FinalApplicationStatus.KILLED, "", ""); am1.unregisterAppAttempt(req, false); am1.addRequests(new String[] { "127.0.0.1" }, GB, 1, 1); AllocateResponse alloc1Response = am1.schedule(); nm1.nodeHeartbeat(true); rm.drainEvents(); alloc1Response = am1.schedule(); Assert.assertEquals(0, alloc1Response.getAllocatedContainers().size()); }
@Test(timeout=60000) public void testInvalidIncreaseDecreaseRequest() throws Exception { conf = new YarnConfiguration(); conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); MockRM rm = new MockRM(conf); try { rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(1024); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); RegisterApplicationMasterResponse registerResponse = am1.registerAppAttempt(); sentRMContainerLaunched(rm, ContainerId.newContainerId(am1.getApplicationAttemptId(), 1)); am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest.newInstance( 0, ContainerId.newContainerId(attempt1.getAppAttemptId(), 1), ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(2048), null))); AllocateResponse response = am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest.newInstance(0, ContainerId.newContainerId(attempt1.getAppAttemptId(), 1), ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(-1), null))); Assert.assertEquals(1, response.getUpdateErrors().size()); Assert.assertEquals("RESOURCE_OUTSIDE_ALLOWED_RANGE", response.getUpdateErrors().get(0).getReason()); response = am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest.newInstance(0, ContainerId.newContainerId(attempt1.getAppAttemptId(), 1), ContainerUpdateType.INCREASE_RESOURCE, Resources.add( registerResponse.getMaximumResourceCapability(), Resources.createResource(1)), null))); Assert.assertEquals(1, response.getUpdateErrors().size()); Assert.assertEquals("RESOURCE_OUTSIDE_ALLOWED_RANGE", response.getUpdateErrors().get(0).getReason()); response = am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest.newInstance(0, ContainerId.newContainerId(attempt1.getAppAttemptId(), 1), ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(2048, 4), null), UpdateContainerRequest.newInstance(0, ContainerId.newContainerId(attempt1.getAppAttemptId(), 1), ContainerUpdateType.DECREASE_RESOURCE, Resources.createResource(1024, 1), null))); Assert.assertEquals(1, response.getUpdateErrors().size()); Assert.assertEquals("UPDATE_OUTSTANDING_ERROR", response.getUpdateErrors().get(0).getReason()); } finally { rm.close(); } }