@Test
  public void testRequestAllocationWithRunStreamProcessor() throws Exception {
    Map<String, String> containersToHostMapping = new HashMap<String, String>() {
      {
        put("0", "prev_host");
        put("1", "prev_host");
        put("2", "prev_host");
        put("3", "prev_host");
      }
    };

    ClusterResourceManager.Callback mockCPM = mock(ClusterResourceManager.Callback.class);
    ClusterResourceManager mockManager = new MockClusterResourceManager(mockCPM, state);
    ContainerManager spyContainerManager =
        spy(new ContainerManager(containerPlacementMetadataStore, state, mockManager, false, false, mock(LocalityManager.class), faultDomainManager, config));
    spyAllocator = Mockito.spy(
        new ContainerAllocator(mockManager, config, state, false, spyContainerManager));
    // Mock the callback from ClusterManager to add resources to the allocator
    doAnswer((InvocationOnMock invocation) -> {
      SamzaResource resource = (SamzaResource) invocation.getArgumentAt(0, List.class).get(0);
      spyAllocator.addResource(resource);
      return null;
    }).when(mockCPM).onResourcesAvailable(anyList());
    // Request Resources
    spyAllocator.requestResources(containersToHostMapping);
    spyThread = new Thread(spyAllocator, "Container Allocator Thread");
    // Start the container allocator thread periodic assignment
    spyThread.start();
    // TODO: we can eliminate the thread sleep if the whole container allocator and test codes are refactored to use
    // a Clock which can be simulated and controlled.
    Thread.sleep(1000);
    // Verify that all the request that were created were "ANY_HOST" requests
    ArgumentCaptor<SamzaResourceRequest> resourceRequestCaptor = ArgumentCaptor.forClass(SamzaResourceRequest.class);
    verify(spyAllocator, times(4)).runStreamProcessor(resourceRequestCaptor.capture(), anyString());
    resourceRequestCaptor.getAllValues()
        .forEach(resourceRequest -> assertEquals(resourceRequest.getPreferredHost(), ResourceRequestState.ANY_HOST));
    assertTrue(state.anyHostRequests.get() == containersToHostMapping.size());
    // Expiry currently should not be invoked for host affinity enabled cases only
    verify(spyContainerManager, never()).handleExpiredRequest(anyString(), anyString(),
        any(SamzaResourceRequest.class), any(ContainerAllocator.class), any(ResourceRequestState.class));
    // Only updated when host affinity is enabled
    assertTrue(state.matchedResourceRequests.get() == 0);
    assertTrue(state.preferredHostRequests.get() == 0);
    spyAllocator.stop();
  }
Test the complete flow from container request creation to allocation when host affinity is disabled
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerAllocatorWithoutHostAffinity.java