/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.samza.clustermanager;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.time.Duration;
import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.samza.SamzaException;
import org.apache.samza.clustermanager.container.placement.ContainerPlacementMetadataStore;
import org.apache.samza.config.ClusterManagerConfig;
import org.apache.samza.config.Config;
import org.apache.samza.config.JobConfig;
import org.apache.samza.config.MetricsConfig;
import org.apache.samza.container.LocalityManager;
import org.apache.samza.container.placement.ContainerPlacementRequestMessage;
import org.apache.samza.diagnostics.DiagnosticsManager;
import org.apache.samza.job.model.ProcessorLocality;
import org.apache.samza.job.model.LocalityModel;
import org.apache.samza.metrics.ContainerProcessManagerMetrics;
import org.apache.samza.metrics.JvmMetrics;
import org.apache.samza.metrics.MetricsRegistryMap;
import org.apache.samza.metrics.MetricsReporter;
import org.apache.samza.metrics.reporter.MetricsSnapshotReporter;
import org.apache.samza.util.DiagnosticsUtil;
import org.apache.samza.util.MetricsReporterLoader;
import org.apache.samza.util.ReflectionUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Option;

import static com.google.common.base.Preconditions.checkNotNull;


/**
 * ContainerProcessManager is responsible for requesting containers, handling failures, and notifying the application master that the
 * job is done.
 *
 * The following threads are involved in the execution of the ContainerProcessManager :
 *  - The main thread (defined in SamzaAppMaster) that sends requests to the cluster manager.
 *  - The callback handler thread that receives the responses from cluster manager and handles:
 *      - Populating a buffer when a container is allocated by the cluster manager
 *        (allocatedContainers in {@link ResourceRequestState}
 *      - Identifying the cause of container failure and re-requesting containers from the cluster manager by adding request to the
 *        internal requestQueue in {@link ResourceRequestState}
 *  - The allocator thread that assigns the allocated containers to pending requests
 *    (See {@link org.apache.samza.clustermanager.ContainerAllocator} or {@link org.apache.samza.clustermanager.ContainerAllocator})
 *
 */
public class ContainerProcessManager implements ClusterResourceManager.Callback   {

  private static final Logger LOG = LoggerFactory.getLogger(ContainerProcessManager.class);

  /**
   * Metrics for the {@link ContainerProcessManager}
   */
  private final static String METRICS_SOURCE_NAME = "ApplicationMaster";
  private final static String EXEC_ENV_CONTAINER_ID_SYS_PROPERTY = "CONTAINER_ID";

  /**
   * Does this Samza Job need hostAffinity when containers are allocated.
   */
  private final boolean hostAffinityEnabled;

  /**
   * State variables tracking containers allocated, freed, running, released.
   */
  private final SamzaApplicationState state;

  /**
   * Config for this Samza job
   */
  private final ClusterManagerConfig clusterManagerConfig;
  private final JobConfig jobConfig;

  /**
   * The Allocator matches requests to resources and executes processes.
   */
  private final ContainerAllocator containerAllocator;
  private final Thread allocatorThread;

  // The ContainerManager manages control actions for both active & standby containers
  private final ContainerManager containerManager;

  private final Option<DiagnosticsManager> diagnosticsManager;

  private final LocalityManager localityManager;

  /**
   * A standard interface to request resources.
   */
  private final ClusterResourceManager clusterResourceManager;

  /**
   * If there are more than job.container.retry.count failures of a container within a job.container.retry.window period,
   * then the ContainerProcessManager will indicate to the ClusterBasedJobCoordinator that the job should shutdown.
   */
  private volatile boolean jobFailureCriteriaMet = false;

  /**
   * Exception thrown in callbacks, such as {@code containerAllocator}
   */
  private volatile Throwable exceptionOccurred = null;

  /**
   * A map that keeps track of how many times each processor failed. The key is the processor ID, and the
   * value is the {@link ProcessorFailure} object that has a count of failures.
   */
  private final Map<String, ProcessorFailure> processorFailures = new HashMap<>();

  private final boolean restartContainers;

  private ContainerProcessManagerMetrics containerProcessManagerMetrics;
  private JvmMetrics jvmMetrics;
  private Map<String, MetricsReporter> metricsReporters;

  public ContainerProcessManager(Config config, SamzaApplicationState state, MetricsRegistryMap registry,
      ContainerPlacementMetadataStore metadataStore, LocalityManager localityManager, boolean restartContainers) {
    Preconditions.checkNotNull(localityManager, "Locality manager cannot be null");
    this.state = state;
    this.clusterManagerConfig = new ClusterManagerConfig(config);
    this.jobConfig = new JobConfig(config);

    this.hostAffinityEnabled = clusterManagerConfig.getHostAffinityEnabled();

    ResourceManagerFactory factory = getContainerProcessManagerFactory(clusterManagerConfig);
    this.clusterResourceManager = checkNotNull(factory.getClusterResourceManager(this, state));

    FaultDomainManagerFactory faultDomainManagerFactory = getFaultDomainManagerFactory(clusterManagerConfig);
    FaultDomainManager faultDomainManager = checkNotNull(faultDomainManagerFactory.getFaultDomainManager(config, registry));

    // Initialize metrics
    this.containerProcessManagerMetrics = new ContainerProcessManagerMetrics(config, state, registry);
    this.jvmMetrics = new JvmMetrics(registry);
    this.metricsReporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), METRICS_SOURCE_NAME);

    // Creating diagnostics manager and reporter, and wiring it respectively
    String jobName = new JobConfig(config).getName().get();
    String jobId = new JobConfig(config).getJobId();
    Optional<String> execEnvContainerId = Optional.ofNullable(System.getenv(EXEC_ENV_CONTAINER_ID_SYS_PROPERTY));
    Optional<Pair<DiagnosticsManager, MetricsSnapshotReporter>> diagnosticsManagerReporterPair =
        DiagnosticsUtil.buildDiagnosticsManager(jobName, jobId, state.jobModelManager.jobModel(), METRICS_SOURCE_NAME, execEnvContainerId, config);

    if (diagnosticsManagerReporterPair.isPresent()) {
      diagnosticsManager = Option.apply(diagnosticsManagerReporterPair.get().getKey());
      metricsReporters.put(MetricsConfig.METRICS_SNAPSHOT_REPORTER_NAME_FOR_DIAGNOSTICS, diagnosticsManagerReporterPair.get().getValue());
    } else {
      diagnosticsManager = Option.empty();
    }

    this.localityManager = localityManager;
    // Wire all metrics to all reporters
    this.metricsReporters.values().forEach(reporter -> reporter.register(METRICS_SOURCE_NAME, registry));

    this.containerManager = new ContainerManager(metadataStore, state, clusterResourceManager,
            hostAffinityEnabled, jobConfig.getStandbyTasksEnabled(), localityManager, faultDomainManager, config);

    this.containerAllocator = new ContainerAllocator(this.clusterResourceManager, config, state, hostAffinityEnabled, this.containerManager);
    this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
    this.restartContainers = restartContainers;
    LOG.info("Finished container process manager initialization.");
  }

  @VisibleForTesting
  ContainerProcessManager(ClusterManagerConfig clusterManagerConfig,
      SamzaApplicationState state,
      MetricsRegistryMap registry,
      ClusterResourceManager resourceManager,
      Optional<ContainerAllocator> allocator,
      ContainerManager containerManager,
      LocalityManager localityManager,
      boolean restartContainers) {
    this.state = state;
    this.clusterManagerConfig = clusterManagerConfig;
    this.jobConfig = new JobConfig(clusterManagerConfig);

    this.hostAffinityEnabled = clusterManagerConfig.getHostAffinityEnabled();

    this.clusterResourceManager = resourceManager;
    this.containerManager = containerManager;
    this.diagnosticsManager = Option.empty();
    this.localityManager = localityManager;
    this.containerAllocator = allocator.orElseGet(
      () -> new ContainerAllocator(this.clusterResourceManager, clusterManagerConfig, state,
          hostAffinityEnabled, this.containerManager));
    this.allocatorThread = new Thread(this.containerAllocator, "Container Allocator Thread");
    this.restartContainers = restartContainers;
    LOG.info("Finished container process manager initialization");
  }

  public boolean shouldShutdown() {
    LOG.debug("ContainerProcessManager state: Completed containers: {}, Configured containers: {}, Are there too many failed containers: {}, Is allocator thread alive: {}",
      state.completedProcessors.get(), state.processorCount, jobFailureCriteriaMet ? "yes" : "no", allocatorThread.isAlive() ? "yes" : "no");

    if (exceptionOccurred != null) {
      LOG.error("Exception in container process manager", exceptionOccurred);
      throw new SamzaException(exceptionOccurred);
    }
    return jobFailureCriteriaMet || state.completedProcessors.get() == state.processorCount.get() || !allocatorThread.isAlive();
  }

  public void start() {
    LOG.info("Starting the container process manager");

    int containerRetryCount = clusterManagerConfig.getContainerRetryCount();
    if (containerRetryCount > -1) {
      LOG.info("Max retries on restarting failed containers: {}", containerRetryCount);
    } else {
      LOG.info("Infinite retries on restarting failed containers");
    }

    if (jvmMetrics != null) {
      jvmMetrics.start();
    }

    if (metricsReporters != null) {
      metricsReporters.values().forEach(reporter -> reporter.start());
    }

    if (diagnosticsManager.isDefined()) {
      diagnosticsManager.get().start();
    }

    // In AM-HA, clusterResourceManager receives already running containers
    // and invokes onStreamProcessorLaunchSuccess which inturn updates state
    // hence state has to be set prior to starting clusterResourceManager.
    state.processorCount.set(state.jobModelManager.jobModel().getContainers().size());
    state.neededProcessors.set(state.jobModelManager.jobModel().getContainers().size());

    LOG.info("Starting the cluster resource manager");
    clusterResourceManager.start();

    // Request initial set of containers
    LocalityModel localityModel = localityManager.readLocality();
    Map<String, String> processorToHost = new HashMap<>();
    state.jobModelManager.jobModel().getContainers().keySet().forEach((processorId) -> {
      String host = Optional.ofNullable(localityModel.getProcessorLocality(processorId))
          .map(ProcessorLocality::host)
          .filter(StringUtils::isNotBlank)
          .orElse(null);
      processorToHost.put(processorId, host);
    });
    if (jobConfig.getApplicationMasterHighAvailabilityEnabled()) {
      // don't request resource for container that is already running
      state.runningProcessors.forEach((processorId, samzaResource) -> {
        LOG.info("Not requesting container for processorId: {} since its already running as containerId: {}",
            processorId, samzaResource.getContainerId());
        processorToHost.remove(processorId);
        if (restartContainers) {
          clusterResourceManager.stopStreamProcessor(samzaResource);
        }
      });
    }
    containerAllocator.requestResources(processorToHost);

    // Start container allocator thread
    LOG.info("Starting the container allocator thread");
    allocatorThread.start();
    LOG.info("Starting the container process manager");
  }

  public void stop() {
    LOG.info("Stopping the container process manager");

    // Shutdown allocator thread
    containerAllocator.stop();
    try {
      allocatorThread.join();
      LOG.info("Stopped container allocator");
    } catch (InterruptedException ie) {
      LOG.error("Allocator thread join threw an interrupted exception", ie);
      Thread.currentThread().interrupt();
    }

    if (diagnosticsManager.isDefined()) {
      try {
        diagnosticsManager.get().stop();
      } catch (InterruptedException e) {
        LOG.error("InterruptedException while stopping diagnosticsManager", e);
      }
    }

    try {

      if (metricsReporters != null) {
        metricsReporters.values().forEach(reporter -> reporter.stop());
      }

      if (jvmMetrics != null) {
        jvmMetrics.stop();
      }

      LOG.info("Stopped containerProcessManagerMetrics reporters");
    } catch (Throwable e) {
      LOG.error("Exception while stopping containerProcessManagerMetrics", e);
    }

    try {
      clusterResourceManager.stop(state.status);
      LOG.info("Stopped the cluster resource manager");
    } catch (Throwable e) {
      LOG.error("Exception while stopping cluster resource manager", e);
    }

    LOG.info("Stopped the container process manager");
  }

  public void onResourceAllocated(SamzaResource resource) {
    LOG.info("Container ID: {} allocated from RM on host: {}", resource.getContainerId(), resource.getHost());
    containerAllocator.addResource(resource);
  }

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

  @Override
  public void onResourcesAvailable(List<SamzaResource> resources) {
    for (SamzaResource resource : resources) {
      onResourceAllocated(resource);
    }
  }

  @Override
  public void onResourcesCompleted(List<SamzaResourceStatus> resourceStatuses) {
    for (SamzaResourceStatus resourceStatus : resourceStatuses) {
      onResourceCompleted(resourceStatus);
    }
  }

  @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);
    }
  }

  @Override
  public void onStreamProcessorLaunchFailure(SamzaResource resource, Throwable t) {
    String containerId = resource.getContainerId();
    String containerHost = resource.getHost();

    // 1. Obtain the pending Samza processor ID for this container ID.
    String processorId = getPendingProcessorId(containerId);
    LOG.error("Launch failed for pending Processor ID: {} on Container ID: {} on host: {} with exception: {}",
        processorId, containerId, containerHost, t);

    // 2. Release resources for containers that failed back to YARN
    LOG.info("Releasing un-startable Container ID: {} for pending Processor ID: {}", containerId, processorId);
    clusterResourceManager.releaseResources(resource);

    // 3. Re-request resources on ANY_HOST in case of launch failures on the preferred host, if standby are not enabled
    // otherwise calling standbyContainerManager
    containerManager.handleContainerLaunchFail(processorId, containerId, containerHost, containerAllocator);
  }

  @Override
  public void onStreamProcessorStopFailure(SamzaResource resource, Throwable t) {
    String containerId = resource.getContainerId();
    String containerHost = resource.getHost();
    String processorId = getRunningProcessor(containerId).getKey();
    LOG.warn("Stop failed for running Processor ID: {} on Container ID: {} on host: {} with exception: {}",
        processorId, containerId, containerHost, t);

    // Notify container-manager of the failed container-stop request
    containerManager.handleContainerStopFail(processorId, containerId, containerHost, containerAllocator);
  }

  /**
   * An error in the callback terminates the JobCoordinator
   * @param e the underlying exception/error
   */
  @Override
  public void onError(Throwable e) {
    LOG.error("Exception occurred in callbacks in the Cluster Resource Manager", e);
    exceptionOccurred = e;
  }

  @VisibleForTesting
  boolean getJobFailureCriteriaMet() {
    return jobFailureCriteriaMet;
  }

  @VisibleForTesting
  Map<String, ProcessorFailure> getProcessorFailures() {
    return processorFailures;
  }

  /**
   * Called within {@link #onResourceCompleted(SamzaResourceStatus)} for unknown exit statuses. These exit statuses
   * correspond to container completion other than container run-to-completion, abort or preemption, or disk failure
   * (e.g., detected by YARN's NM healthchecks).
   * @param resourceStatus reported resource status.
   * @param containerId container ID
   * @param processorId processor ID (aka. logical container ID)
   * @param exitStatus exit status from the {@link #onResourceCompleted(SamzaResourceStatus)} callback.
   */
  @VisibleForTesting
  void onResourceCompletedWithUnknownStatus(SamzaResourceStatus resourceStatus, String containerId, String processorId,
      int exitStatus) {
    LOG.info("Container ID: {} for Processor ID: {} failed with exit code: {}.", containerId, processorId, exitStatus);
    Instant now = Instant.now();
    state.failedContainers.incrementAndGet();
    state.jobHealthy.set(false);

    state.neededProcessors.incrementAndGet();
    // Find out previously running container location
    String lastSeenOn = Optional.ofNullable(localityManager.readLocality().getProcessorLocality(processorId))
        .map(ProcessorLocality::host)
        .orElse(null);
    if (!hostAffinityEnabled || StringUtils.isBlank(lastSeenOn)) {
      lastSeenOn = ResourceRequestState.ANY_HOST;
    }
    LOG.info("Container ID: {} for Processor ID: {} was last seen on host {}.", containerId, processorId, lastSeenOn);
    // A container failed for an unknown reason. Let's check to see if
    // we need to shutdown the whole app master if too many container
    // failures have happened. The rules for failing are that the
    // failure count for a task group id must be > the configured retry
    // count, and the last failure (the one prior to this one) must have
    // happened less than retry window ms ago. If retry count is set to
    // 0, the app master will fail on any container failure. If the
    // retry count is set to a number < 0, a container failure will
    // never trigger an app master failure.
    int retryCount = clusterManagerConfig.getContainerRetryCount();
    int retryWindowMs = clusterManagerConfig.getContainerRetryWindowMs();
    int currentFailCount;
    boolean retryContainerRequest = true;

    if (retryCount == 0) {
      // Failure criteria met only if failed containers can fail the job.
      jobFailureCriteriaMet = clusterManagerConfig.shouldFailJobAfterContainerRetries();
      if (jobFailureCriteriaMet) {
        LOG.error("Processor ID: {} (current Container ID: {}) failed, and retry count is set to 0, " +
            "so shutting down the application master and marking the job as failed.", processorId, containerId);
      } else {
        LOG.error("Processor ID: {} (current Container ID: {}) failed, and retry count is set to 0, " +
            "but the job will continue to run with the failed container.", processorId, containerId);
        state.failedProcessors.put(processorId, resourceStatus);
      }
      retryContainerRequest = false;
    } else if (retryCount > 0) {
      long durationSinceLastRetryMs;
      if (processorFailures.containsKey(processorId)) {
        ProcessorFailure failure = processorFailures.get(processorId);
        currentFailCount = failure.getCount() + 1;
        Duration lastRetryDelay = getRetryDelay(processorId);
        Instant retryAttemptedAt = failure.getLastFailure().plus(lastRetryDelay);
        durationSinceLastRetryMs = now.toEpochMilli() - retryAttemptedAt.toEpochMilli();
        if (durationSinceLastRetryMs < 0) {
          // This should never happen without changes to the system clock or time travel. Log a warning just in case.
          LOG.warn("Last failure at: {} with a retry attempted at: {} which is supposed to be before current time of: {}",
              failure.getLastFailure(), retryAttemptedAt, now);
        }
      } else {
        currentFailCount = 1;
        durationSinceLastRetryMs = 0;
      }

      if (durationSinceLastRetryMs >= retryWindowMs) {
        LOG.info("Resetting failure count for Processor ID: {} back to 1, since last failure " +
            "(for Container ID: {}) was outside the bounds of the retry window.", processorId, containerId);

        // Reset counter back to 1, since the last failure for this
        // container happened outside the window boundary.
        currentFailCount = 1;
      }

      // if fail count is (1 initial failure + max retries) then fail job.
      if (currentFailCount > retryCount) {
        LOG.error("Processor ID: {} (current Container ID: {}) has failed {} times, with last failure {} ms ago. " +
                "This is greater than retry count of {} and window of {} ms, ",
            processorId, containerId, currentFailCount, durationSinceLastRetryMs, retryCount, retryWindowMs);

        // We have too many failures, and we're within the window
        // boundary, so reset shut down the app master.
        retryContainerRequest = false;
        if (clusterManagerConfig.shouldFailJobAfterContainerRetries()) {
          jobFailureCriteriaMet = true;
          LOG.error("Shutting down the application master and marking the job as failed after max retry attempts.");
          state.status = SamzaApplicationState.SamzaAppStatus.FAILED;
        } else {
          LOG.warn("Processor ID: {} with Container ID: {} failed after all retry attempts. Job will continue to run without this container.",
              processorId, containerId);
          state.failedProcessors.put(processorId, resourceStatus);
        }
      } else {
        LOG.info("Current failure count for Processor ID: {} is {}.", processorId, currentFailCount);
        Duration retryDelay = Duration.ZERO;
        if (!ResourceRequestState.ANY_HOST.equals(lastSeenOn) && currentFailCount == retryCount) {
          // Add the preferred host last retry delay on the last retry
          retryDelay = Duration.ofMillis(clusterManagerConfig.getContainerPreferredHostLastRetryDelayMs());
        }
        processorFailures.put(processorId, new ProcessorFailure(currentFailCount, now, retryDelay));
        retryContainerRequest = true;
      }
    }

    if (retryContainerRequest) {
      Duration retryDelay = getRetryDelay(processorId);
      if (!retryDelay.isZero()) {
        LOG.info("Adding a delay of: {} seconds on the last container retry request for preferred host: {}",
            retryDelay.getSeconds(), lastSeenOn);
      }
      handleContainerStop(processorId, resourceStatus.getContainerId(), lastSeenOn, exitStatus, retryDelay);
    }
  }

  /**
   * Registers a ContainerPlacement action, this method is invoked by ContainerPlacementRequestAllocator. {@link ContainerProcessManager}
   * needs to intercept container placement actions between ContainerPlacementRequestAllocator and {@link ContainerManager} to avoid
   * cyclic dependency between {@link ContainerManager} and {@link ContainerAllocator} on each other
   *
   * @param requestMessage request containing details of the desited container placement action
   */
  public void registerContainerPlacementAction(ContainerPlacementRequestMessage requestMessage) {
    containerManager.registerContainerPlacementAction(requestMessage, containerAllocator);
  }

  private Duration getRetryDelay(String processorId) {
    return processorFailures.containsKey(processorId)
        ? processorFailures.get(processorId).getLastRetryDelay()
        : Duration.ZERO;
  }

  /**
   * Returns an instantiated {@link ResourceManagerFactory} from a {@link ClusterManagerConfig}. The
   * {@link ResourceManagerFactory} is used to return an implementation of a {@link ClusterResourceManager}
   *
   * @param clusterManagerConfig, the cluster manager config to parse.
   *
   */
  private ResourceManagerFactory getContainerProcessManagerFactory(final ClusterManagerConfig clusterManagerConfig) {
    final String containerManagerFactoryClass = clusterManagerConfig.getContainerManagerClass();
    final ResourceManagerFactory factory;

    try {
      factory = ReflectionUtil.getObj(containerManagerFactoryClass, ResourceManagerFactory.class);
    } catch (Exception e) {
      LOG.error("Error creating the cluster resource manager.", e);
      throw new SamzaException(e);
    }
    return factory;
  }

  /**
   * Returns an instantiated {@link FaultDomainManagerFactory} from a {@link ClusterManagerConfig}. The
   * {@link FaultDomainManagerFactory} is used to return an implementation of a {@link FaultDomainManager}
   *
   * @param clusterManagerConfig, the cluster manager config to parse.
   *
   */
  private FaultDomainManagerFactory getFaultDomainManagerFactory(final ClusterManagerConfig clusterManagerConfig) {
    final String faultDomainManagerFactoryClass = clusterManagerConfig.getFaultDomainManagerClass();
    final FaultDomainManagerFactory factory;

    try {
      factory = ReflectionUtil.getObj(faultDomainManagerFactoryClass, FaultDomainManagerFactory.class);
    } catch (Exception e) {
      LOG.error("Error creating the fault domain manager.", e);
      throw new SamzaException(e);
    }
    return factory;
  }

  /**
   * Obtains the ID of the Samza processor pending launch on the provided resource (container).
   *
   * ContainerProcessManager [INFO] Container ID: container_e66_1569376389369_0221_01_000049 matched pending Processor ID: 0 on host: ltx1-app0772.stg.linkedin.com
   *
   * @param containerId last known id of the container deployed
   * @return the logical processorId of the processor (e.g., 0, 1, 2 ...)
   */
  private String getPendingProcessorId(String containerId) {
    for (Map.Entry<String, SamzaResource> entry: state.pendingProcessors.entrySet()) {
      if (entry.getValue().getContainerId().equals(containerId)) {
        LOG.info("Container ID: {} matched pending Processor ID: {} on host: {}", containerId, entry.getKey(), entry.getValue().getHost());
        return entry.getKey();
      }
    }
    return null;
  }

  private Pair<String, String> getRunningProcessor(String containerId) {
    for (Map.Entry<String, SamzaResource> entry: state.runningProcessors.entrySet()) {
      if (entry.getValue().getContainerId().equals(containerId)) {
        LOG.info("Container ID: {} matched running Processor ID: {} on host: {}", containerId, entry.getKey(), entry.getValue().getHost());

        String processorId = entry.getKey();
        String hostName = entry.getValue().getHost();
        return new ImmutablePair<>(processorId, hostName);
      }
    }

    return new ImmutablePair<>(null, null);
  }

  /**
   * Request {@link ContainerManager#handleContainerStop} to determine next step of actions for the stopped container
   */
  private void handleContainerStop(String processorId, String containerId, String preferredHost, int exitStatus, Duration preferredHostRetryDelay) {
    containerManager.handleContainerStop(processorId, containerId, preferredHost, exitStatus, preferredHostRetryDelay, containerAllocator);
  }
}