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.isPresent()) {
      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");
  }
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.
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerProcessManager.java