protected void runStreamProcessor(SamzaResourceRequest request, String preferredHost) {
    CommandBuilder builder = getCommandBuilder(request.getProcessorId());
    // Get the available resource
    SamzaResource resource = peekAllocatedResource(preferredHost);
    if (resource == null) {
      throw new SamzaException("Expected resource for Processor ID: " + request.getProcessorId() + " was unavailable on host: " + preferredHost);
    }

    /**
     * If the allocated resource has expired then release the expired resource and re-request the resources from {@link ClusterResourceManager}
     */
    if (clusterResourceManager.isResourceExpired(resource)) {
      containerManager.handleExpiredResource(request, resource, preferredHost, resourceRequestState, this);
      return;
    }

    // Update state
    resourceRequestState.updateStateAfterAssignment(request, preferredHost, resource);
    String processorId = request.getProcessorId();

    // Run processor on resource
    LOG.info("Found Container ID: {} for Processor ID: {} on host: {} for request creation time: {}.",
        resource.getContainerId(), processorId, preferredHost, request.getRequestTimestamp());

    // Update processor state as "pending" and then issue a request to launch it. It's important to perform the state-update
    // prior to issuing the request. Otherwise, there's a race where the response callback may arrive sooner and not see
    // the processor as "pending" (SAMZA-2117)

    state.failedProcessors.remove(processorId);
    state.pendingProcessors.put(processorId, resource);

    clusterResourceManager.launchStreamProcessor(resource, builder);
  }
Updates the request state and runs a processor on the specified host. Assumes a resource
is available on the preferred host, so the caller must verify that before invoking this method.

@param request             the {@link SamzaResourceRequest} which is being handled.
@param preferredHost       the preferred host on which the processor should be run or
                           {@link ResourceRequestState#ANY_HOST} if there is no host preference.
@throws                    SamzaException if there is no allocated resource in the specified host.
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerAllocator.java