@Private
  public void scheduleAllReduces() {
    for (ContainerRequest req : pendingReduces) {
      scheduledRequests.addReduce(req);
    }
    pendingReduces.clear();
  }
@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 = MockRMAppSubmitter.submitWithMemory(2048, rm);
    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());

    // request to allocate two reduce priority containers
    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();
    }
    // only 1 allocated container should be assigned
    assertThat(assignedContainer).isEqualTo(1);
  }