boolean handleContainerLaunch(SamzaResourceRequest request, String preferredHost, SamzaResource allocatedResource,
      ResourceRequestState resourceRequestState, ContainerAllocator allocator) {
    if (hasActiveContainerPlacementAction(request.getProcessorId())) {
      String processorId = request.getProcessorId();
      ContainerPlacementMetadata actionMetaData = getPlacementActionMetadata(processorId).get();
      ContainerPlacementMetadata.ContainerStatus actionStatus = actionMetaData.getContainerStatus();
      if (samzaApplicationState.runningProcessors.containsKey(processorId) && actionStatus == ContainerPlacementMetadata.ContainerStatus.RUNNING) {
        LOG.debug("Requesting running container to shutdown due to existing ContainerPlacement action {}", actionMetaData);
        actionMetaData.setContainerStatus(ContainerPlacementMetadata.ContainerStatus.STOP_IN_PROGRESS);
        updateContainerPlacementActionStatus(actionMetaData, ContainerPlacementMessage.StatusCode.IN_PROGRESS,
            "Active container stop in progress");
        clusterResourceManager.stopStreamProcessor(samzaApplicationState.runningProcessors.get(processorId));
        return false;
      } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOP_IN_PROGRESS) {
        LOG.info("Waiting for running container to shutdown due to existing ContainerPlacement action {}", actionMetaData);
        return false;
      } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOP_FAILED) {
        LOG.info("Shutdown on running container failed for action {}", actionMetaData);
        markContainerPlacementActionFailed(actionMetaData,
            String.format("failed to stop container on current host %s", actionMetaData.getSourceHost()));
        resourceRequestState.cancelResourceRequest(request);
        return true;
      } else if (actionStatus == ContainerPlacementMetadata.ContainerStatus.STOPPED) {
        // If the job has standby containers enabled, always check standby constraints before issuing a start on container
        // Note: Always check constraints against allocated resource, since preferred host can be ANY_HOST as well
        if (standbyContainerManager.isPresent() && !standbyContainerManager.get().checkStandbyConstraints(request.getProcessorId(), allocatedResource.getHost())) {
          LOG.info(
              "Starting container {} on host {} does not meet standby constraints, falling back to source host placement metadata: {}",
              request.getProcessorId(), preferredHost, actionMetaData);
          // Release unstartable container
          standbyContainerManager.get().releaseUnstartableContainer(request, allocatedResource, preferredHost, resourceRequestState);
          // Fallback to source host since the new allocated resource does not meet standby constraints
          allocator.requestResource(processorId, actionMetaData.getSourceHost());
          markContainerPlacementActionFailed(actionMetaData,
              String.format("allocated resource %s does not meet standby constraints now, falling back to source host", allocatedResource));
        } else {
          LOG.info("Status updated for ContainerPlacement action: ", actionMetaData);
          allocator.runStreamProcessor(request, preferredHost);
        }
        return true;
      }
    }

    if (this.standbyContainerManager.isPresent()) {
      standbyContainerManager.get()
          .checkStandbyConstraintsAndRunStreamProcessor(request, preferredHost, allocatedResource, allocator,
              resourceRequestState);
    } else {
      allocator.runStreamProcessor(request, preferredHost);
    }
    return true;
  }
Handles the container start action for both active & standby containers. This method is invoked by the allocator thread

Case 1. If the container launch request is due to an existing container placement action, issue a stop on active
        container & wait for the active container to be stopped before issuing a start.
Case 2. If StandbyContainer is present refer to {@code StandbyContainerManager#checkStandbyConstraintsAndRunStreamProcessor}
Case 3. Otherwise just invoke a container start on the allocated resource for the pending request

TODO: SAMZA-2399: Investigate & configure a timeout for container stop if needed

@param request pending request for the preferred host
@param preferredHost preferred host to start the container
@param allocatedResource resource allocated from {@link ClusterResourceManager}
@param resourceRequestState state of request in {@link ContainerAllocator}
@param allocator to request resources from @{@link ClusterResourceManager}

@return true if the container launch is complete, false if the container launch is in progress. A container launch
        might be in progress when it is waiting for the previous container incarnation to stop in case of container
        placement actions
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java