public void onResourceCompleted(SamzaResourceStatus resourceStatus) {
    String containerId = resourceStatus.getContainerId();
    Pair<String, String> runningProcessorIdHostname = getRunningProcessor(containerId);
    String processorId = runningProcessorIdHostname.getKey();
    String hostName = runningProcessorIdHostname.getValue();

    if (processorId == null) {
      LOG.info("No running Processor ID found for Container ID: {} with Status: {}. Ignoring redundant notification.", containerId, resourceStatus.toString());
      state.redundantNotifications.incrementAndGet();

      if (resourceStatus.getExitCode() != SamzaResourceStatus.SUCCESS) {
        // the requested container failed before assigning the request to it.
        // Remove from the buffer if it is there
        containerAllocator.releaseResource(containerId);
      }
      return;
    }
    state.runningProcessors.remove(processorId);
    int exitStatus = resourceStatus.getExitCode();
    switch (exitStatus) {
      case SamzaResourceStatus.SUCCESS:
        LOG.info("Container ID: {} for Processor ID: {} completed successfully.", containerId, processorId);

        state.completedProcessors.incrementAndGet();

        state.finishedProcessors.incrementAndGet();
        processorFailures.remove(processorId);

        if (state.completedProcessors.get() == state.processorCount.get()) {
          LOG.info("Setting job status to SUCCEEDED since all containers have been marked as completed.");
          state.status = SamzaApplicationState.SamzaAppStatus.SUCCEEDED;
        }
        break;

      case SamzaResourceStatus.DISK_FAIL:
      case SamzaResourceStatus.ABORTED:
      case SamzaResourceStatus.PREEMPTED:
        LOG.info("Container ID: {} for Processor ID: {} was released with an exit code: {}. This means that " +
                "the container was killed by YARN, either due to being released by the application master " +
                "or being 'lost' due to node failures etc. or due to preemption by the RM." +
                "Requesting a new container for the processor.",
                containerId, processorId, exitStatus);

        state.releasedContainers.incrementAndGet();

        // If this container was assigned some partitions (a processorId), then
        // clean up, and request a new container for the processor. This only
        // should happen if the container was 'lost' due to node failure, not
        // if the AM released the container.
        state.neededProcessors.incrementAndGet();
        state.jobHealthy.set(false);

        // handle container stop due to node fail
        handleContainerStop(processorId, resourceStatus.getContainerId(), ResourceRequestState.ANY_HOST, exitStatus, Duration.ZERO);
        break;

      default:
        onResourceCompletedWithUnknownStatus(resourceStatus, containerId, processorId, exitStatus);
    }

    if (diagnosticsManager.isPresent()) {
      diagnosticsManager.get().addProcessorStopEvent(processorId, resourceStatus.getContainerId(), hostName, exitStatus);
    }
  }
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