@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(); }
private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; }
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 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); }
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(); }
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; }