@Test
  public void testContainerAssignment() throws Exception {
    // Host-affinity enabled
    ResourceRequestState state = new ResourceRequestState(true, manager);
    SamzaResourceRequest request = new SamzaResourceRequest(1, 1024, "abc", "0");

    SamzaResourceRequest request1 = new SamzaResourceRequest(1, 1024, "def", "0");

    state.addResourceRequest(request);
    state.addResourceRequest(request1);

    SamzaResource container = new SamzaResource(1, 1024, "abc", "id0");

    SamzaResource container1 = new SamzaResource(1, 1024, "zzz", "id1");
    state.addResource(container);
    state.addResource(container1);

    assertEquals(2, state.numPendingRequests());
    assertEquals(2, state.getHostRequestCounts().size());

    assertNotNull(state.getResourcesOnAHost("abc"));
    assertEquals(1, state.getResourcesOnAHost("abc").size());
    assertEquals(container, state.getResourcesOnAHost("abc").get(0));

    assertNotNull(state.getResourcesOnAHost("def"));
    assertEquals(0, state.getResourcesOnAHost("def").size());

    assertNotNull(state.getResourcesOnAHost(ANY_HOST));
    assertEquals(1, state.getResourcesOnAHost(ANY_HOST).size());
    assertEquals(container1, state.getResourcesOnAHost(ANY_HOST).get(0));

    // Container assigned on the requested host
    state.updateStateAfterAssignment(request, "abc", container);

    assertEquals(request1, state.peekPendingRequest());

    assertNotNull(state.getHostRequestCounts().get("abc"));
    assertEquals(0, state.getHostRequestCounts().get("abc").get());

    assertNotNull(state.getResourcesOnAHost("abc"));
    assertEquals(0, state.getResourcesOnAHost("abc").size());

    // Container assigned on any host
    state.updateStateAfterAssignment(request1, ANY_HOST, container1);

    assertEquals(0, state.numPendingRequests());

    assertNotNull(state.getHostRequestCounts().get("def"));
    assertEquals(0, state.getHostRequestCounts().get("def").get());

    assertNotNull(state.getResourcesOnAHost(ANY_HOST));
    assertEquals(0, state.getResourcesOnAHost(ANY_HOST).size());

  }
Test request state after container is assigned to a host
* Assigned on requested host
* Assigned on any host
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerRequestState.java