public void setupStandby() throws Exception {
    LocalityManager mockLocalityManager = mock(LocalityManager.class);
    when(mockLocalityManager.readLocality())
        .thenReturn(new LocalityModel(ImmutableMap.of(
            "0", new ProcessorLocality("0", "host-1"),
            "1", new ProcessorLocality("1", "host-2"),
            "0-0", new ProcessorLocality("0", "host-2"),
            "1-0", new ProcessorLocality("0", "host-1"))));
    state = new SamzaApplicationState(getJobModelManagerWithStandby());
    callback = mock(ClusterResourceManager.Callback.class);
    MockClusterResourceManager clusterResourceManager = new MockClusterResourceManager(callback, state);
    FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
    ClusterManagerConfig clusterManagerConfig = new ClusterManagerConfig(config);
    // Enable standby
    containerManager = spy(new ContainerManager(containerPlacementMetadataStore, state, clusterResourceManager, true, true, mockLocalityManager, faultDomainManager, config));
    allocatorWithHostAffinity = new MockContainerAllocatorWithHostAffinity(clusterResourceManager, config, state, containerManager);
    cpm = new ContainerProcessManager(clusterManagerConfig, state, new MetricsRegistryMap(),
        clusterResourceManager, Optional.of(allocatorWithHostAffinity), containerManager, mockLocalityManager, false);
  }
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