@Test
  public void testExpiredRequestsAreCancelled() throws Exception {
    // request one container each on host-1 and host-2
    containerAllocator.requestResources(ImmutableMap.of("0", "host-1", "1", "host-2"));
    // assert that the requests made it to YARN
    Assert.assertEquals(clusterResourceManager.resourceRequests.size(), 2);
    // allocate one resource from YARN on a different host (host-3)
    SamzaResource resource0 = new SamzaResource(1, 1000, "host-3", "id1");
    containerAllocator.addResource(resource0);
    // let the matching begin
    allocatorThread.start();

    // verify that a container is launched on host-3 after the request expires
    if (!clusterResourceManager.awaitContainerLaunch(1, 20, TimeUnit.SECONDS)) {
      Assert.fail("Timed out waiting container launch");
    }
    Assert.assertEquals(1, clusterResourceManager.launchedResources.size());
    Assert.assertEquals(clusterResourceManager.launchedResources.get(0).getHost(), "host-3");
    Assert.assertEquals(clusterResourceManager.launchedResources.get(0).getContainerId(), "id1");

    // Now, there are no more resources left to run the 2nd container. Verify that we eventually issue another request
    if (!clusterResourceManager.awaitResourceRequests(4, 20, TimeUnit.SECONDS)) {
      Assert.fail("Timed out waiting for resource requests");
    }
    // verify that we have cancelled previous requests and there's one outstanding request
    Assert.assertEquals(clusterResourceManager.cancelledRequests.size(), 3);
  }
Handles expired requests correctly and assigns ANY_HOST
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java