@Test(timeout = 10000)
  public void testAlwaysMoveToAnyHostForHostAffinityDisabled() throws Exception {
    Map<String, String> conf = new HashMap<>();
    conf.putAll(getConfigWithHostAffinityAndRetries(false, 1, true));
    SamzaApplicationState state =
        new SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(), 2, this.server));
    ClusterResourceManager.Callback callback = mock(ClusterResourceManager.Callback.class);
    MockClusterResourceManager clusterResourceManager = new MockClusterResourceManager(callback, state);
    FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
    ContainerManager containerManager =
        new ContainerManager(containerPlacementMetadataStore, state, clusterResourceManager, false, false, localityManager, faultDomainManager, config);
    MockContainerAllocatorWithoutHostAffinity allocatorWithoutHostAffinity =
        new MockContainerAllocatorWithoutHostAffinity(clusterResourceManager, new MapConfig(conf), state,
            containerManager);

    ContainerProcessManager cpm = new ContainerProcessManager(
        new ClusterManagerConfig(new MapConfig(getConfig(), getConfigWithHostAffinityAndRetries(false, 1, true))), state,
        new MetricsRegistryMap(), clusterResourceManager, Optional.of(allocatorWithoutHostAffinity), containerManager, localityManager, false);

    // Mimic Cluster Manager returning any request
    doAnswer(new Answer<Void>() {
      public Void answer(InvocationOnMock invocation) {
        new Thread(() -> {
          Object[] args = invocation.getArguments();
          List<SamzaResource> resources = (List<SamzaResource>) args[0];
          SamzaResource preferredResource = resources.get(0);
          SamzaResource anyResource =
              new SamzaResource(preferredResource.getNumCores(), preferredResource.getMemoryMb(),
                  "host-" + RandomStringUtils.randomAlphanumeric(5), preferredResource.getContainerId());
          cpm.onResourcesAvailable(ImmutableList.of(anyResource));
        }, "AMRMClientAsync").start();
        return null;
      }
    }).when(callback).onResourcesAvailable(anyList());

    doAnswer(new Answer<Void>() {
      public Void answer(InvocationOnMock invocation) {
          new Thread(() -> {
            Object[] args = invocation.getArguments();
            cpm.onStreamProcessorLaunchSuccess((SamzaResource) args[0]);
          }, "AMRMClientAsync").start();
        return null;
      }
    }).when(callback).onStreamProcessorLaunchSuccess(any());

    doAnswer(new Answer<Void>() {
      public Void answer(InvocationOnMock invocation) {
          new Thread(() -> {
            Object[] args = invocation.getArguments();
            cpm.onResourcesCompleted((List<SamzaResourceStatus>) args[0]);
          }, "AMRMClientAsync").start();
        return null;
      }
    }).when(callback).onResourcesCompleted(anyList());

    cpm.start();

    // This spawns async start request and waits for async requests to complete
    if (!allocatorWithoutHostAffinity.awaitContainersStart(2, 3, TimeUnit.SECONDS)) {
      fail("timed out waiting for the containers to start");
    }

    while (state.runningProcessors.size() != 2) {
      Thread.sleep(100);
    }

    // App is in running state with two containers running
    assertEquals(state.runningProcessors.size(), 2);
    assertEquals(state.preferredHostRequests.get(), 0);
    assertEquals(state.anyHostRequests.get(), 2);

    String previousHostOfContainer1 = state.runningProcessors.get("0").getHost();
    String previousHostOfContainer2 = state.runningProcessors.get("1").getHost();

    // Initiate container placement action to move a container with container id 0
    ContainerPlacementRequestMessage requestMessage =
        new ContainerPlacementRequestMessage(UUID.randomUUID(), "app-attempt-001", "0", "host-3",
            System.currentTimeMillis());
    ContainerPlacementMetadata metadata =
        containerManager.registerContainerPlacementActionForTest(requestMessage, allocatorWithoutHostAffinity);

    // Wait for the ControlAction to complete and spawn an async request
    if (!allocatorWithoutHostAffinity.awaitContainersStart(1, 3, TimeUnit.SECONDS)) {
      fail("timed out waiting for the containers to start");
    }

    Optional<ContainerPlacementResponseMessage> responseMessage =
        containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestMessage.getUuid());

    while (true) {
      if (metadata.getActionStatus() == ContainerPlacementMessage.StatusCode.SUCCEEDED && responseMessage.isPresent()
          && responseMessage.get().getStatusCode() == ContainerPlacementMessage.StatusCode.SUCCEEDED) {
        break;
      }
      Thread.sleep(100);
      responseMessage = containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestMessage.getUuid());
    }

    // We should have no preferred host request
    assertEquals(0, state.preferredHostRequests.get());
    // We should have one more ANY_HOST request
    assertEquals(3, state.anyHostRequests.get());
    assertEquals(2, state.runningProcessors.size());
    assertNotEquals(previousHostOfContainer1, state.runningProcessors.get("0").getHost());
    // Container 2 should not be affected
    assertEquals(previousHostOfContainer2, state.runningProcessors.get("1").getHost());
    assertEquals(3, state.anyHostRequests.get());
    // Action should success
    assertEquals(ContainerPlacementMessage.StatusCode.SUCCEEDED, metadata.getActionStatus());

    assertTrue(responseMessage.isPresent());
    assertEquals(responseMessage.get().getStatusCode(), ContainerPlacementMessage.StatusCode.SUCCEEDED);
    assertResponseMessage(responseMessage.get(), requestMessage);

    /**
     * Inject a duplicate request and check it is not accepted
     */
    ContainerPlacementRequestMessage duplicateRequestToBeIgnored =
        new ContainerPlacementRequestMessage(requestMessage.getUuid(), "app-attempt-001", "1",
            "host-3", System.currentTimeMillis());

    // Request with a dup uuid should not be accepted
    metadata = containerManager.registerContainerPlacementActionForTest(duplicateRequestToBeIgnored,
        allocatorWithoutHostAffinity);
    // metadata should be from the previous completed action
    assertTrue(metadata == null || metadata.getUuid() != duplicateRequestToBeIgnored.getUuid());

    responseMessage =
        containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestMessage.getUuid());

    assertTrue(responseMessage.isPresent());
    assertEquals(responseMessage.get().getStatusCode(), ContainerPlacementMessage.StatusCode.BAD_REQUEST);
    assertResponseMessage(responseMessage.get(), duplicateRequestToBeIgnored);

    // Request shall be deleted as soon as it is acted upon
    assertFalse(containerPlacementMetadataStore.readContainerPlacementRequestMessage(requestMessage.getUuid()).isPresent());
  }
Set of Integration tests for container placement actions

Please note that semaphores are used wherever possible, there are some Thread.sleep used for the main thread to check
on state changes to atomic variables or synchroized metadata objects because of difficulty of plugging semaphores to
those pieces of logic
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerPlacementActions.java