private ContainerProcessManager buildContainerProcessManager(ClusterManagerConfig clusterManagerConfig, SamzaApplicationState state,
      ClusterResourceManager clusterResourceManager, Optional<ContainerAllocator> allocator, LocalityManager localityManager,
      boolean restartContainers, FaultDomainManager faultDomainManager) {
    return new ContainerProcessManager(clusterManagerConfig, state, new MetricsRegistryMap(), clusterResourceManager,
        allocator, buildContainerManager(containerPlacementMetadataStore, state, clusterResourceManager,
        clusterManagerConfig.getHostAffinityEnabled(), false, localityManager, faultDomainManager), localityManager, restartContainers);
  }
Test AM requests a new container when a task fails
Error codes with same behavior - Disk failure, preemption and aborted
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java