@Test(timeout = 20000)
  public void testContainerPlacementsForJobRunningInDegradedState() throws Exception {
    // Set failure after retries to false to enable job running in degraded state
    config = new MapConfig(configVals, getConfigWithHostAffinityAndRetries(true, 1, false));
    state = new SamzaApplicationState(JobModelManagerTestUtil.getJobModelManager(getConfig(), 2, this.server));
    callback = mock(ClusterResourceManager.Callback.class);
    MockClusterResourceManager clusterResourceManager = new MockClusterResourceManager(callback, state);
    FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
    ClusterManagerConfig clusterManagerConfig = new ClusterManagerConfig(config);
    containerManager = spy(new ContainerManager(containerPlacementMetadataStore, state, clusterResourceManager, true, false, localityManager, faultDomainManager, config));
    allocatorWithHostAffinity = new MockContainerAllocatorWithHostAffinity(clusterResourceManager, config, state, containerManager);
    cpm = new ContainerProcessManager(clusterManagerConfig, state, new MetricsRegistryMap(),
        clusterResourceManager, Optional.of(allocatorWithHostAffinity), containerManager, localityManager, false);

    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-2,
    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();


    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);


    // Trigger a container failure
    clusterResourceManager.stopStreamProcessor(state.runningProcessors.get("1"), -103);
    // Wait for container to start
    if (!allocatorWithHostAffinity.awaitContainersStart(1, 2, TimeUnit.SECONDS)) {
      fail("timed out waiting for the containers to start");
    }
    while (state.runningProcessors.size() != 2) {
      Thread.sleep(100);
    }
    // Trigger a container failure again
    clusterResourceManager.stopStreamProcessor(state.runningProcessors.get("1"), -103);
    // Ensure that this container has exhausted all retires
    while (state.failedProcessors.size() != 1 && state.runningProcessors.size() != 1) {
      Thread.sleep(100);
    }

    // At this point the application should only have one container running
    assertEquals(state.runningProcessors.get("0").getHost(), "host-1");
    assertEquals(state.runningProcessors.size(), 1);
    assertEquals(state.pendingProcessors.size(), 0);
    assertTrue(state.failedProcessors.containsKey("1"));

    ContainerPlacementRequestMessage requestMessage =
        new ContainerPlacementRequestMessage(UUID.randomUUID(), "app-attempt-001", "1", "host-3",
            System.currentTimeMillis());

    ContainerPlacementMetadata metadata =
        containerManager.registerContainerPlacementActionForTest(requestMessage, allocatorWithHostAffinity);

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

    // Wait for both the containers to be in running state & control action metadata to succeed
    while (state.runningProcessors.size() != 2
        && metadata.getActionStatus() != ContainerPlacementMessage.StatusCode.SUCCEEDED) {
      Thread.sleep(100);
    }

    assertEquals(state.preferredHostRequests.get(), 4);
    assertEquals(state.runningProcessors.size(), 2);
    // Container 1 should not go to host-3
    assertEquals(state.runningProcessors.get("0").getHost(), "host-1");
    assertEquals(state.runningProcessors.get("1").getHost(), "host-3");
    assertEquals(state.anyHostRequests.get(), 0);
    // Failed processors must be empty
    assertEquals(state.failedProcessors.size(), 0);
  }
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