@Test
  public void testAddContainer() {
    // Add container to ANY_LIST when host-affinity is not enabled
    ResourceRequestState state = new ResourceRequestState(false, manager);
    SamzaResource resource = new SamzaResource(1, 1024, "abc", "id1");

    state.addResource(resource);

    assertNotNull(state.getHostRequestCounts());
    assertNotNull(state.getResourcesOnAHost(ANY_HOST));

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

    // Container Allocated when there is no request in queue
    ResourceRequestState state1 = spy(new ResourceRequestState(true, manager));
    SamzaResource container1 = new SamzaResource(1, 1024, "zzz", "id2");
    state1.addResource(container1);

    assertEquals(0, state1.numPendingRequests());

    assertNull(state1.getResourcesOnAHost("zzz"));
    assertNotNull(state1.getResourcesOnAHost(ANY_HOST));
    assertEquals(1, state1.getResourcesOnAHost(ANY_HOST).size());
    assertEquals(container1, state1.getResourcesOnAHost(ANY_HOST).get(0));

    // Container Allocated on a Requested Host
    state1.addResourceRequest(new SamzaResourceRequest(1, 1024, "abc", "0"));

    // Delayed request
    state1.addResourceRequest(new SamzaResourceRequest(1, 1024, "def", "1",
        Instant.now().plus(Duration.ofHours(1))));
    state1.addResourceRequest(new SamzaResourceRequest(1, 1024, "ghi", "2",
        Instant.now().plus(Duration.ofHours(2))));

    assertEquals(1, state1.numPendingRequests());
    assertEquals(2, state1.numDelayedRequests());

    // Verify request sent only once for the non-delayed request
    verify(state1).sendResourceRequest(any(SamzaResourceRequest.class));

    assertNotNull(state1.getHostRequestCounts());
    assertNotNull(state1.getHostRequestCounts().get("abc"));
    assertEquals(1, state1.getHostRequestCounts().get("abc").get());

    state1.addResource(resource);

    assertNotNull(state1.getResourcesOnAHost("abc"));
    assertEquals(1, state1.getResourcesOnAHost("abc").size());
    assertEquals(resource, state1.getResourcesOnAHost("abc").get(0));

    // Container Allocated on host that was not requested
    SamzaResource container2 = new SamzaResource(1, 1024, "xyz", "id2");

    state1.addResource(container2);

    assertNull(state1.getResourcesOnAHost("xyz"));
    assertNotNull(state1.getResourcesOnAHost(ANY_HOST));
    assertEquals(2, state1.getResourcesOnAHost(ANY_HOST).size());
    assertEquals(container2, state1.getResourcesOnAHost(ANY_HOST).get(1));

    // Extra containers were allocated on a host that was requested
    SamzaResource container3 = new SamzaResource(1, 1024, "abc", "id3");
    state1.addResource(container3);

    assertEquals(3, state1.getResourcesOnAHost(ANY_HOST).size());
    assertEquals(container3, state1.getResourcesOnAHost(ANY_HOST).get(2));
  }
Test addContainer() updates the state correctly
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerRequestState.java