@Override
  public void onStreamProcessorLaunchSuccess(SamzaResource resource) {
    String containerId = resource.getContainerId();
    String containerHost = resource.getHost();

    // 1. Obtain the processor ID for the pending container on this resource.
    String processorId = getPendingProcessorId(containerId);
    LOG.info("Successfully started Processor ID: {} on Container ID: {} on host: {}",
        processorId, containerId, containerHost);

    // 2. Remove the container from the pending buffer and add it to the running buffer. Additionally, update the
    // job-health metric.
    if (processorId != null) {
      LOG.info("Moving Processor ID: {} on Container ID: {} on host: {} from pending to running state.",
          processorId, containerId, containerHost);
      state.pendingProcessors.remove(processorId);
      state.runningProcessors.put(processorId, resource);
      if (state.neededProcessors.decrementAndGet() == 0) {
        state.jobHealthy.set(true);
      }
      containerManager.handleContainerLaunchSuccess(processorId, containerHost);
    } else {
      LOG.warn("Did not find a pending Processor ID for Container ID: {} on host: {}. " +
          "Ignoring invalid/redundant notification.", containerId, containerHost);
    }
  }
This methods handles the onResourceCompleted callback from the RM. Based on the ContainerExitStatus, it decides
whether the container that exited is marked as complete or failure.
@param resourceStatus status of the resource that completed
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java