RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { @Private public void scheduleAllReduces() { for (ContainerRequest req : pendingReduces) { scheduledRequests.addReduce(req); } pendingReduces.clear(); } RMContainerAllocator(ClientService clientService, AppContext context,
      AMPreemptionPolicy preemptionPolicy); 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(); static final float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART; @VisibleForTesting
  static final String RAMPDOWN_DIAGNOSTIC; }
@Test(timeout = 30000) public void testExcessReduceContainerAssign() throws Exception { final Configuration conf = new Configuration(); conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 0.0f); final MyResourceManager2 rm = new MyResourceManager2(conf); rm.start(); final RMApp app = rm.submitApp(2048); rm.drainEvents(); final String host = "host1"; final MockNM nm = rm.registerNode(String.format("%s:1234", host), 4096); nm.nodeHeartbeat(true); rm.drainEvents(); final ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); rm.drainEvents(); final JobId jobId = MRBuilderUtils .newJobId(appAttemptId.getApplicationId(), 0); final 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, "")); final MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob, SystemClock.getInstance()); final String[] locations = new String[] {host}; allocator.sendRequest(createRequest(jobId, 0, Resource.newInstance(1024, 1), locations, false, true)); allocator.scheduleAllReduces(); allocator.makeRemoteRequest(); nm.nodeHeartbeat(true); rm.drainEvents(); allocator.sendRequest(createRequest(jobId, 1, Resource.newInstance(1024, 1), locations, false, false)); int assignedContainer; for (assignedContainer = 0; assignedContainer < 1;) { assignedContainer += allocator.schedule().size(); nm.nodeHeartbeat(true); rm.drainEvents(); } Assert.assertEquals(assignedContainer, 1); }
@Test public void testUnsupportedReduceContainerRequirement() throws Exception { final Resource maxContainerSupported = Resource.newInstance(1, 1); final ApplicationId appId = ApplicationId.newInstance(1, 1); final ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance( appId, 1); final JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); final MockScheduler mockScheduler = new MockScheduler(appAttemptId); final Configuration conf = new Configuration(); final MyContainerAllocator allocator = new MyContainerAllocator(null, conf, appAttemptId, mock(Job.class), SystemClock.getInstance()) { @Override protected void register() { } @Override protected ApplicationMasterProtocol createSchedulerProxy() { return mockScheduler; } @Override protected Resource getMaxContainerCapability() { return maxContainerSupported; } }; final int memory = (int) (maxContainerSupported.getMemorySize() + 10); ContainerRequestEvent reduceRequestEvt = createRequest(jobId, 0, Resource.newInstance(memory, maxContainerSupported.getVirtualCores()), new String[0], false, true); allocator.sendRequests(Arrays.asList(reduceRequestEvt)); allocator.scheduleAllReduces(); allocator.schedule(); Assert.assertEquals(0, mockScheduler.lastAnyAskReduce); }
@Test(timeout = 30000) public void testExcessReduceContainerAssign() throws Exception { final Configuration conf = new Configuration(); conf.setFloat(MRJobConfig.COMPLETED_MAPS_FOR_REDUCE_SLOWSTART, 0.0f); final MyResourceManager2 rm = new MyResourceManager2(conf); rm.start(); final DrainDispatcher dispatcher = (DrainDispatcher)rm.getRMContext() .getDispatcher(); final RMApp app = rm.submitApp(2048); dispatcher.await(); final String host = "host1"; final MockNM nm = rm.registerNode(String.format("%s:1234", host), 4096); nm.nodeHeartbeat(true); dispatcher.await(); final ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); final JobId jobId = MRBuilderUtils .newJobId(appAttemptId.getApplicationId(), 0); final 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, "")); final MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob, new SystemClock()); final String[] locations = new String[] { host }; allocator.sendRequest(createReq(jobId, 0, 1024, locations, false, true)); allocator.scheduleAllReduces(); allocator.makeRemoteRequest(); nm.nodeHeartbeat(true); dispatcher.await(); allocator.sendRequest(createReq(jobId, 1, 1024, locations, false, false)); int assignedContainer; for (assignedContainer = 0; assignedContainer < 1;) { assignedContainer += allocator.schedule().size(); nm.nodeHeartbeat(true); dispatcher.await(); } Assert.assertEquals(assignedContainer, 1); }