RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; } RMContainerAllocator(ClientService clientService, AppContext context); boolean getIsReduceStarted(); void setIsReduceStarted(boolean reduceStarted); @Override void handle(ContainerAllocatorEvent event); @Private void scheduleReduces(
      int totalMaps, int completedMaps,
      int scheduledMaps, int scheduledReduces,
      int assignedMaps, int assignedReduces,
      Resource mapResourceReqt, Resource reduceResourceReqt,
      int numPendingReduces,
      float maxReduceRampupLimit, float reduceSlowStart); @Private void scheduleAllReduces(); @Private void rampUpReduces(int rampUp); @Private void rampDownReduces(int rampDown); @VisibleForTesting TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptID); @Private Resource getResourceLimit(); @VisibleForTesting int getNumOfPendingReduces(); static final float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART; @VisibleForTesting
  static final String RAMPDOWN_DIAGNOSTIC; }
@Test public void testMapNodeLocality() throws Exception { LOG.info("Running testMapNodeLocality"); Configuration conf = new Configuration(); MyResourceManager rm = new MyResourceManager(conf); rm.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext() .getDispatcher(); RMApp app = rm.submitApp(1024); dispatcher.await(); MockNM amNodeManager = rm.registerNode("amNM:1234", 2048); amNodeManager.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob); MockNM nodeManager1 = rm.registerNode("h1:1234", 3072); rm.registerNode("h2:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 1536); dispatcher.await(); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 1024, new String[] { "h1" }); allocator.sendRequest(event2); ContainerRequestEvent event3 = createReq(jobId, 3, 1024, new String[] { "h2" }); allocator.sendRequest(event3); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); nodeManager3.nodeHeartbeat(true); nodeManager1.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); checkAssignments(new ContainerRequestEvent[] { event1, event2, event3 }, assigned, false); for(TaskAttemptContainerAssignedEvent event : assigned) { if(event.getTaskAttemptID().equals(event3.getAttemptID())) { assigned.remove(event); Assert.assertTrue( event.getContainer().getNodeId().getHost().equals("h3")); break; } } checkAssignments(new ContainerRequestEvent[] { event1, event2}, assigned, true); }
@Test public void testMapReduceScheduling() throws Exception { LOG.info("Running testMapReduceScheduling"); Configuration conf = new Configuration(); MyResourceManager rm = new MyResourceManager(conf); rm.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext() .getDispatcher(); RMApp app = rm.submitApp(1024); dispatcher.await(); MockNM amNodeManager = rm.registerNode("amNM:1234", 2048); amNodeManager.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob, new SystemClock()); MockNM nodeManager1 = rm.registerNode("h1:1234", 1024); MockNM nodeManager2 = rm.registerNode("h2:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 10240); dispatcher.await(); ContainerRequestEvent event1 = createReq(jobId, 1, 2048, new String[] { "h1", "h2" }, true, false); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 3000, new String[] { "h1" }, false, true); allocator.sendRequest(event2); ContainerRequestEvent event3 = createReq(jobId, 3, 2048, new String[] { "h3" }, false, false); allocator.sendRequest(event3); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); nodeManager1.nodeHeartbeat(true); nodeManager2.nodeHeartbeat(true); nodeManager3.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); checkAssignments(new ContainerRequestEvent[] { event1, event3 }, assigned, false); for (TaskAttemptContainerAssignedEvent assig : assigned) { Assert.assertFalse("Assigned count not correct", "h1".equals(assig .getContainer().getNodeId().getHost())); } }
@Test public void testBlackListedNodes() throws Exception { LOG.info("Running testBlackListedNodes"); Configuration conf = new Configuration(); conf.setBoolean(MRJobConfig.MR_AM_JOB_NODE_BLACKLISTING_ENABLE, true); conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 1); conf.setInt( MRJobConfig.MR_AM_IGNORE_BLACKLISTING_BLACKLISTED_NODE_PERECENT, -1); MyResourceManager rm = new MyResourceManager(conf); rm.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext() .getDispatcher(); RMApp app = rm.submitApp(1024); dispatcher.await(); MockNM amNodeManager = rm.registerNode("amNM:1234", 2048); amNodeManager.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob); MockNM nodeManager1 = rm.registerNode("h1:1234", 10240); MockNM nodeManager2 = rm.registerNode("h2:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 10240); dispatcher.await(); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 1024, new String[] { "h2" }); allocator.sendRequest(event2); ContainerRequestEvent event3 = createReq(jobId, 3, 1024, new String[] { "h3" }); allocator.sendRequest(event3); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); ContainerFailedEvent f1 = createFailEvent(jobId, 1, "h1", false); allocator.sendFailure(f1); ContainerFailedEvent f2 = createFailEvent(jobId, 1, "h2", false); allocator.sendFailure(f2); nodeManager1.nodeHeartbeat(true); nodeManager2.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); assertBlacklistAdditionsAndRemovals(2, 0, rm); nodeManager1.nodeHeartbeat(false); nodeManager2.nodeHeartbeat(false); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); nodeManager3.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertTrue("No of assignments must be 3", assigned.size() == 3); for (TaskAttemptContainerAssignedEvent assig : assigned) { Assert.assertTrue("Assigned container host not correct", "h3".equals(assig .getContainer().getNodeId().getHost())); } }
@Test public void testBlackListedNodesWithSchedulingToThatNode() throws Exception { LOG.info("Running testBlackListedNodesWithSchedulingToThatNode"); Configuration conf = new Configuration(); conf.setBoolean(MRJobConfig.MR_AM_JOB_NODE_BLACKLISTING_ENABLE, true); conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 1); conf.setInt( MRJobConfig.MR_AM_IGNORE_BLACKLISTING_BLACKLISTED_NODE_PERECENT, -1); MyResourceManager rm = new MyResourceManager(conf); rm.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext() .getDispatcher(); RMApp app = rm.submitApp(1024); dispatcher.await(); MockNM amNodeManager = rm.registerNode("amNM:1234", 2048); amNodeManager.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob); MockNM nodeManager1 = rm.registerNode("h1:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 10240); dispatcher.await(); LOG.info("Requesting 1 Containers _1 on H1"); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); LOG.info("RM Heartbeat (to send the container requests)"); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); LOG.info("h1 Heartbeat (To actually schedule the containers)"); nodeManager1.nodeHeartbeat(true); dispatcher.await(); LOG.info("RM Heartbeat (To process the scheduled containers)"); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertEquals("No of assignments must be 1", 1, assigned.size()); LOG.info("Failing container _1 on H1 (should blacklist the node)"); ContainerFailedEvent f1 = createFailEvent(jobId, 1, "h1", false); allocator.sendFailure(f1); ContainerRequestEvent event1f = createReq(jobId, 1, 1024, new String[] { "h1" }, true, false); allocator.sendRequest(event1f); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(1, 0, rm); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); ContainerRequestEvent event3 = createReq(jobId, 3, 1024, new String[] { "h1", "h3" }); allocator.sendRequest(event3); LOG.info("h1 Heartbeat (To actually schedule the containers)"); nodeManager1.nodeHeartbeat(true); dispatcher.await(); LOG.info("RM Heartbeat (To process the scheduled containers)"); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); LOG.info("RM Heartbeat (To process the re-scheduled containers)"); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); LOG.info("h3 Heartbeat (To re-schedule the containers)"); nodeManager3.nodeHeartbeat(true); dispatcher.await(); LOG.info("RM Heartbeat (To process the re-scheduled containers for H3)"); assigned = allocator.schedule(); assertBlacklistAdditionsAndRemovals(0, 0, rm); dispatcher.await(); for (TaskAttemptContainerAssignedEvent assig : assigned) { LOG.info(assig.getTaskAttemptID() + " assgined to " + assig.getContainer().getId() + " with priority " + assig.getContainer().getPriority()); } Assert.assertEquals("No of assignments must be 2", 2, assigned.size()); for (TaskAttemptContainerAssignedEvent assig : assigned) { Assert.assertEquals("Assigned container " + assig.getContainer().getId() + " host not correct", "h3", assig.getContainer().getNodeId().getHost()); } }
@Test public void testRMContainerAllocatorResendsRequestsOnRMRestart() throws Exception { Configuration conf = new Configuration(); conf.set(YarnConfiguration.RECOVERY_ENABLED, "true"); conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName()); conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, YarnConfiguration.DEFAULT_RM_AM_MAX_ATTEMPTS); conf.setBoolean(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_ENABLED, true); conf.setLong(YarnConfiguration.RM_WORK_PRESERVING_RECOVERY_SCHEDULING_WAIT_MS, 0); conf.setBoolean(MRJobConfig.MR_AM_JOB_NODE_BLACKLISTING_ENABLE, true); conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 1); conf.setInt( MRJobConfig.MR_AM_IGNORE_BLACKLISTING_BLACKLISTED_NODE_PERECENT, -1); MemoryRMStateStore memStore = new MemoryRMStateStore(); memStore.init(conf); MyResourceManager rm1 = new MyResourceManager(conf, memStore); rm1.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm1.getRMContext().getDispatcher(); RMApp app = rm1.submitApp(1024); dispatcher.await(); MockNM nm1 = new MockNM("h1:1234", 15120, rm1.getResourceTrackerService()); nm1.registerNode(); nm1.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt().getAppAttemptId(); rm1.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm1, conf, appAttemptId, mockJob); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 2048, new String[] { "h1", "h2" }); allocator.sendRequest(event2); ContainerFailedEvent f1 = createFailEvent(jobId, 1, "h2", false); allocator.sendFailure(f1); List<TaskAttemptContainerAssignedEvent> assignedContainers = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assignedContainers.size()); assertAsksAndReleases(3, 0, rm1); assertBlacklistAdditionsAndRemovals(1, 0, rm1); nm1.nodeHeartbeat(true); dispatcher.await(); assignedContainers = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 2", 2, assignedContainers.size()); assertAsksAndReleases(0, 0, rm1); assertBlacklistAdditionsAndRemovals(0, 0, rm1); assignedContainers = allocator.schedule(); Assert.assertEquals("No of assignments must be 0", 0, assignedContainers.size()); assertAsksAndReleases(3, 0, rm1); assertBlacklistAdditionsAndRemovals(0, 0, rm1); ContainerRequestEvent event3 = createReq(jobId, 3, 1000, new String[] { "h1" }); allocator.sendRequest(event3); ContainerAllocatorEvent deallocate1 = createDeallocateEvent(jobId, 1, false); allocator.sendDeallocate(deallocate1); assignedContainers = allocator.schedule(); Assert.assertEquals("No of assignments must be 0", 0, assignedContainers.size()); assertAsksAndReleases(3, 1, rm1); assertBlacklistAdditionsAndRemovals(0, 0, rm1); MyResourceManager rm2 = new MyResourceManager(conf, memStore); rm2.start(); nm1.setResourceTrackerService(rm2.getResourceTrackerService()); allocator.updateSchedulerProxy(rm2); dispatcher = (DrainDispatcher) rm2.getRMContext().getDispatcher(); NodeHeartbeatResponse hbResponse = nm1.nodeHeartbeat(true); Assert.assertEquals(NodeAction.RESYNC, hbResponse.getNodeAction()); nm1 = new MockNM("h1:1234", 10240, rm2.getResourceTrackerService()); nm1.registerNode(); nm1.nodeHeartbeat(true); dispatcher.await(); ContainerAllocatorEvent deallocate2 = createDeallocateEvent(jobId, 2, false); allocator.sendDeallocate(deallocate2); ContainerFailedEvent f2 = createFailEvent(jobId, 1, "h3", false); allocator.sendFailure(f2); ContainerRequestEvent event4 = createReq(jobId, 4, 2000, new String[] { "h1", "h2" }); allocator.sendRequest(event4); allocator.schedule(); dispatcher.await(); ContainerRequestEvent event5 = createReq(jobId, 5, 3000, new String[] { "h1", "h2", "h3" }); allocator.sendRequest(event5); assignedContainers = allocator.schedule(); dispatcher.await(); assertAsksAndReleases(3, 2, rm2); assertBlacklistAdditionsAndRemovals(2, 0, rm2); nm1.nodeHeartbeat(true); dispatcher.await(); assignedContainers = allocator.schedule(); dispatcher.await(); Assert.assertEquals("Number of container should be 3", 3, assignedContainers.size()); for (TaskAttemptContainerAssignedEvent assig : assignedContainers) { Assert.assertTrue("Assigned count not correct", "h1".equals(assig.getContainer().getNodeId().getHost())); } rm1.stop(); rm2.stop(); }