public void testExpiryWithNonResponsiveClusterManager() throws Exception {

    final SamzaResource resource0 = new SamzaResource(1, 1000, "host-3", "id1");
    final SamzaResource resource1 = new SamzaResource(1, 1000, "host-4", "id2");

    Map<String, String> containersToHostMapping = ImmutableMap.of("0", "host-1", "1", "host-2");

    Runnable addContainerAssertions = new Runnable() {
      @Override
      public void run() {
        // verify that resources are buffered in the right queue. ie, only those resources on previously requested hosts
        // in the preferred-host queue while other resources end up in the ANY_HOST queue
        assertNull(requestState.getResourcesOnAHost("host-3"));
        assertNull(requestState.getResourcesOnAHost("host-4"));
        assertNotNull(requestState.getResourcesOnAHost(ResourceRequestState.ANY_HOST));
        assertEquals(1, requestState.getResourcesOnAHost(ResourceRequestState.ANY_HOST).size());
      }
    };

    Runnable assignContainerAssertions = new Runnable() {
      // verify that we processed all requests
      @Override
      public void run() {
        assertEquals(requestState.numPendingRequests(), 0);
      }
    };

    Runnable runningContainerAssertions = new Runnable() {
      // verify that the two containers were actually launched
      @Override
      public void run() {
        assertTrue(clusterResourceManager.launchedResources.contains(resource0));
        assertTrue(clusterResourceManager.launchedResources.contains(resource1));
      }
    };
    MockContainerListener listener = new MockContainerListener(2, 0, 2, 2, addContainerAssertions, null, assignContainerAssertions, runningContainerAssertions);
    requestState.registerContainerListener(listener);
    clusterResourceManager.registerContainerListener(listener);

    // request for resources - one each on host-1 and host-2
    containerAllocator.requestResources(containersToHostMapping);
    assertEquals(requestState.numPendingRequests(), 2);
    assertNotNull(requestState.getHostRequestCounts());
    assertNotNull(requestState.getHostRequestCounts().get("host-1"));
    assertTrue(requestState.getHostRequestCounts().get("host-1").get() == 1);
    assertNotNull(requestState.getHostRequestCounts().get("host-2"));
    assertTrue(requestState.getHostRequestCounts().get("host-2").get() == 1);

    // verify that no containers have been launched yet (since, YARN has not provided any resources)
    assertEquals(0, clusterResourceManager.launchedResources.size());

    // provide a resource on host-3
    containerAllocator.addResource(resource0);
    allocatorThread.start();
    // verify that the first preferred host request should expire and container-0 should launch on host-3
    if (!clusterResourceManager.awaitContainerLaunch(1, 20, TimeUnit.SECONDS)) {
      Assert.fail("Timed out waiting for container-0 to launch");
    }
    // verify that the second preferred host request should expire and should trigger ANY_HOST requests
    // wait for 4 requests to be made (2 preferred-host requests - one each on host-1 & host-2;  2 any-host requests)
    if (!clusterResourceManager.awaitResourceRequests(4, 20, TimeUnit.SECONDS)) {
      Assert.fail("Timed out waiting for resource requests");
    }
    // verify 2 preferred host requests should have been made for host-1 and host-2
    Assert.assertEquals(2, state.preferredHostRequests.get());
    // verify both of them should have expired.
    Assert.assertEquals(2, state.expiredPreferredHostRequests.get());
    // verify there were at-least 2 any-host requests
    Assert.assertTrue(state.anyHostRequests.get() >= 2);
    Assert.assertTrue(state.expiredAnyHostRequests.get() <= state.anyHostRequests.get());
    // finally, provide a container from YARN after multiple requests
    containerAllocator.addResource(resource1);
    // verify all the test assertions
    listener.verify();
  }
Handles expired requests correctly and assigns ANY_HOST
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java