@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);
}