@Test(timeout = 30000)
  public void testActionQueuingForConsecutivePlacementActions() throws Exception {
    // Spawn a Request Allocator Thread
    ContainerPlacementRequestAllocator requestAllocator =
        new ContainerPlacementRequestAllocator(containerPlacementMetadataStore, cpm, new ApplicationConfig(config), 100);
    Thread requestAllocatorThread = new Thread(requestAllocator, "ContainerPlacement Request Allocator Thread");

    requestAllocatorThread.start();

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

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

    // Initiate container placement action to move a container with container id 0

    UUID requestUUIDMove1 = containerPlacementMetadataStore.writeContainerPlacementRequestMessage("appAttempt-001", "0", "host-3",
        null, System.currentTimeMillis());

    UUID requestUUIDMoveBad = containerPlacementMetadataStore.writeContainerPlacementRequestMessage("appAttempt-002", "0", "host-4",
        null, System.currentTimeMillis());

    UUID requestUUIDMove2 = containerPlacementMetadataStore.writeContainerPlacementRequestMessage("appAttempt-001", "0", "host-4",
        null, System.currentTimeMillis());

    // Wait for the ControlAction to complete
    while (true) {
      if (containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestUUIDMove2).isPresent() &&
          containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestUUIDMove2).get().getStatusCode()
              == ContainerPlacementMessage.StatusCode.SUCCEEDED) {
        break;
      }
      Thread.sleep(100);
    }

    assertEquals(state.preferredHostRequests.get(), 4);
    assertEquals(state.runningProcessors.size(), 2);
    assertEquals(state.runningProcessors.get("0").getHost(), "host-4");
    assertEquals(state.runningProcessors.get("1").getHost(), "host-2");
    assertEquals(state.anyHostRequests.get(), 0);

    Optional<ContainerPlacementResponseMessage> responseMessageMove1 =
        containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestUUIDMove1);

    Optional<ContainerPlacementResponseMessage> responseMessageMove2 =
        containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestUUIDMove2);

    assertTrue(responseMessageMove1.isPresent());
    assertEquals(responseMessageMove1.get().getStatusCode(), ContainerPlacementMessage.StatusCode.SUCCEEDED);

    assertTrue(responseMessageMove2.isPresent());
    assertEquals(responseMessageMove2.get().getStatusCode(), ContainerPlacementMessage.StatusCode.SUCCEEDED);

    // Request should be deleted as soon as ita accepted / being acted upon
    assertFalse(containerPlacementMetadataStore.readContainerPlacementRequestMessage(requestUUIDMove1).isPresent());
    assertFalse(containerPlacementMetadataStore.readContainerPlacementRequestMessage(requestUUIDMove2).isPresent());

    // Requests from Previous deploy must be cleaned
    assertFalse(containerPlacementMetadataStore.readContainerPlacementRequestMessage(requestUUIDMoveBad).isPresent());
    assertFalse(containerPlacementMetadataStore.readContainerPlacementResponseMessage(requestUUIDMoveBad).isPresent());

    // Cleanup Request Allocator Thread
    cleanUpRequestAllocatorThread(requestAllocator, requestAllocatorThread);
  }
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