@Test(timeout = 10000)
  public void testActiveContainerLaunchFailureOnControlActionShouldFallbackToSourceHost() throws Exception {
    doAnswer(new Answer<Void>() {
      public Void answer(InvocationOnMock invocation) {
        new Thread(() -> {
          Object[] args = invocation.getArguments();
          cpm.onResourcesAvailable((List<SamzaResource>) args[0]);
        }, "AMRMClientAsync").start();
        return null;
      }
    }).when(callback).onResourcesAvailable(anyList());

    // Mimic stream processor launch failure only on host-3
    doAnswer(new Answer<Void>() {
      public Void answer(InvocationOnMock invocation) {
        new Thread(() -> {
          Object[] args = invocation.getArguments();
          SamzaResource host3Resource = (SamzaResource) args[0];
          if (host3Resource.getHost().equals("host-3")) {
            cpm.onStreamProcessorLaunchFailure(host3Resource, new Throwable("Custom Exception for Host-3"));
          } else {
            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();

    if (!allocatorWithHostAffinity.awaitContainersStart(2, 5, 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.runningProcessors.get("0").getHost(), "host-1");
    assertEquals(state.runningProcessors.get("1").getHost(), "host-2");
    assertEquals(state.preferredHostRequests.get(), 2);
    assertEquals(state.anyHostRequests.get(), 0);

    // Take a 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, allocatorWithHostAffinity);

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

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

    assertEquals(state.preferredHostRequests.get(), 4);
    assertEquals(state.runningProcessors.size(), 2);
    // Container 0 should fallback to source host
    assertEquals(state.runningProcessors.get("0").getHost(), "host-1");
    assertEquals(state.runningProcessors.get("1").getHost(), "host-2");
    assertEquals(state.anyHostRequests.get(), 0);
    // Control Action should be failed in this case
    assertEquals(metadata.getActionStatus(), ContainerPlacementMessage.StatusCode.FAILED);

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

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

    // 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