@Test
  public void testExpiredRequestAllocationOnSurplusAnyHostWithRunStreamProcessor() throws Exception {
    // Add Extra Resources
    MockClusterResourceManager spyClusterResourceManager = spy(new MockClusterResourceManager(callback, state));
    ContainerManager spyContainerManager =
        spy(new ContainerManager(containerPlacementMetadataStore, state, spyClusterResourceManager, true, false, mock(LocalityManager.class), faultDomainManager, config));

    spyAllocator = Mockito.spy(
        new ContainerAllocator(spyClusterResourceManager, config, state, true, spyContainerManager));
    spyAllocator.addResource(new SamzaResource(1, 1000, "xyz", "id1"));
    spyAllocator.addResource(new SamzaResource(1, 1000, "zzz", "id2"));

    // Request Preferred Resources
    spyAllocator.requestResources(new HashMap<String, String>() {
      {
        put("0", "hostname-0");
        put("1", "hostname-1");
      }
    });

    spyAllocatorThread = new Thread(spyAllocator);
    // Start the container allocator thread periodic assignment
    spyAllocatorThread.start();

    // Let the request expire, expiration timeout is 500 ms
    Thread.sleep(1000);

    // Verify that all the request that were created as preferred host requests expired
    assertEquals(state.expiredPreferredHostRequests.get(), 2);
    verify(spyContainerManager, times(1)).handleExpiredRequest(eq("0"), eq("hostname-0"),
        any(SamzaResourceRequest.class), any(ContainerAllocator.class), any(ResourceRequestState.class));
    verify(spyContainerManager, times(1)).handleExpiredRequest(eq("1"), eq("hostname-1"),
        any(SamzaResourceRequest.class), any(ContainerAllocator.class), any(ResourceRequestState.class));

    // Verify that runStreamProcessor was invoked with already available ANY_HOST requests
    ArgumentCaptor<SamzaResourceRequest> resourceRequestCaptor = ArgumentCaptor.forClass(SamzaResourceRequest.class);
    ArgumentCaptor<String> hostCaptor = ArgumentCaptor.forClass(String.class);
    verify(spyAllocator, times(2)).runStreamProcessor(resourceRequestCaptor.capture(), hostCaptor.capture());
    // Resource request were preferred host requests
    resourceRequestCaptor.getAllValues()
        .forEach(resourceRequest -> assertNotEquals(resourceRequest.getPreferredHost(), ResourceRequestState.ANY_HOST));

    // Since requests expired, allocator ran the requests on surplus available ANY_HOST
    hostCaptor.getAllValues()
        .forEach(host -> assertEquals(host, ResourceRequestState.ANY_HOST));

    // State Update check
    assertTrue(state.matchedResourceRequests.get() == 0);
    assertTrue(state.preferredHostRequests.get() == 2);
    assertTrue(state.anyHostRequests.get() == 0);
    spyAllocator.stop();
  }
Handles expired requests correctly and assigns ANY_HOST
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithHostAffinity.java