@Test
  public void testRequestContainersWithExistingHosts() throws Exception {
    Map<String, String> containersToHostMapping = new HashMap<String, String>() {
      {
        put("0", "prev_host");
        put("1", "prev_host");
        put("2", "prev_host");
        put("3", "prev_host");
      }
    };

    allocatorThread.start();

    containerAllocator.requestResources(containersToHostMapping);

    assertEquals(4, manager.resourceRequests.size());

    assertNotNull(requestState);

    assertEquals(4, requestState.numPendingRequests());
    assertEquals(0, requestState.numDelayedRequests());

    // If host-affinty is not enabled, it doesn't update the requestMap
    assertNotNull(requestState.getHostRequestCounts());
    assertEquals(0, requestState.getHostRequestCounts().keySet().size());

    assertNotNull(state);
    assertEquals(4, state.anyHostRequests.get());
    assertEquals(0, state.preferredHostRequests.get());
  }
Test requestContainers with containerToHostMapping with host.affinity disabled
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java