private void testContainerRequestedRetriesNotExceedingWindowOnFailureWithUnknownCode(boolean withHostAffinity, boolean failAfterRetries) throws Exception {
    int maxRetries = 3;
    String processorId = "0";
    ClusterManagerConfig clusterManagerConfig = new ClusterManagerConfig(getConfigWithHostAffinityAndRetries(withHostAffinity, maxRetries, failAfterRetries));
    SamzaApplicationState state = new SamzaApplicationState(getJobModelManager(1));
    MockClusterResourceManagerCallback callback = new MockClusterResourceManagerCallback();
    MockClusterResourceManager clusterResourceManager = new MockClusterResourceManager(callback, state);
    FaultDomainManager faultDomainManager = mock(FaultDomainManager.class);
    LocalityManager mockLocalityManager = mock(LocalityManager.class);

    if (withHostAffinity) {
      when(mockLocalityManager.readLocality())
          .thenReturn(new LocalityModel(ImmutableMap.of("0", new ProcessorLocality("0", "host1"))));
    } else {
      when(mockLocalityManager.readLocality())
          .thenReturn(new LocalityModel(new HashMap<>()));
    }

    ContainerManager containerManager =
        buildContainerManager(containerPlacementMetadataStore, state, clusterResourceManager,
            clusterManagerConfig.getHostAffinityEnabled(), false, mockLocalityManager, faultDomainManager);

    MockContainerAllocatorWithoutHostAffinity allocator = new MockContainerAllocatorWithoutHostAffinity(
        clusterResourceManager,
        clusterManagerConfig,
        state,
        containerManager);

    ContainerProcessManager cpm =
        buildContainerProcessManager(clusterManagerConfig, state, clusterResourceManager, Optional.of(allocator),
            mockLocalityManager, false, faultDomainManager);

    // start triggers a request
    cpm.start();

    assertFalse(cpm.shouldShutdown());
    assertEquals(1, allocator.getContainerRequestState().numPendingRequests());
    assertEquals(0, allocator.getContainerRequestState().numDelayedRequests());

    SamzaResource container = new SamzaResource(1, 1024, "host1", "id0");
    cpm.onResourceAllocated(container);

    // Allow container to run and update state
    if (!allocator.awaitContainersStart(1, 2, TimeUnit.SECONDS)) {
      fail("timed out waiting for the containers to start");
    }
    cpm.onStreamProcessorLaunchSuccess(container);

    // Mock 2nd failure not exceeding retry window.
    cpm.getProcessorFailures().put(processorId, new ProcessorFailure(1, Instant.now(), Duration.ZERO));
    cpm.onResourceCompleted(new SamzaResourceStatus(container.getContainerId(), "diagnostics", 1));
    assertEquals(false, cpm.getJobFailureCriteriaMet());
    assertEquals(2, cpm.getProcessorFailures().get(processorId).getCount());
    assertFalse(cpm.shouldShutdown());
    assertEquals(1, allocator.getContainerRequestState().numPendingRequests());
    assertEquals(0, allocator.getContainerRequestState().numDelayedRequests());

    cpm.onResourceAllocated(container);

    // Allow container to run and update state
    if (!allocator.awaitContainersStart(1, 2, TimeUnit.SECONDS)) {
      fail("timed out waiting for the containers to start");
    }
    cpm.onStreamProcessorLaunchSuccess(container);

    // Mock 3rd failure not exceeding retry window.
    cpm.getProcessorFailures().put(processorId, new ProcessorFailure(2, Instant.now(), Duration.ZERO));
    cpm.onResourceCompleted(new SamzaResourceStatus(container.getContainerId(), "diagnostics", 1));
    assertEquals(false, cpm.getJobFailureCriteriaMet());
    assertEquals(3, cpm.getProcessorFailures().get(processorId).getCount());
    assertFalse(cpm.shouldShutdown());

    if (withHostAffinity) {
      assertEquals(0, allocator.getContainerRequestState().numPendingRequests());
      assertEquals(1, allocator.getContainerRequestState().numDelayedRequests());
    } else {
      assertEquals(1, allocator.getContainerRequestState().numPendingRequests());
      assertEquals(0, allocator.getContainerRequestState().numDelayedRequests());
    }

    cpm.onResourceAllocated(container);

    if (withHostAffinity) {
      if (allocator.awaitContainersStart(1, 2, TimeUnit.SECONDS)) {
        // No delayed retry requests for there host affinity is disabled. Call back should return immediately.
        fail("Expecting a delayed request so allocator callback should have timed out waiting for a response.");
      }

      // For the sake of testing the mocked 4th failure below, send delayed requests now.
      SamzaResourceRequest request = allocator.getContainerRequestState().getDelayedRequestsQueue().poll();
      SamzaResourceRequest fastForwardRequest =
          new SamzaResourceRequest(request.getNumCores(), request.getMemoryMB(), request.getPreferredHost(), request.getProcessorId(), Instant.now().minusSeconds(1));
      allocator.getContainerRequestState().getDelayedRequestsQueue().add(fastForwardRequest);
      int numSent = allocator.getContainerRequestState().sendPendingDelayedResourceRequests();
      assertEquals(1, numSent);
      cpm.onResourceAllocated(container);
    }

    if (!allocator.awaitContainersStart(1, 2, TimeUnit.SECONDS)) {
      // No delayed retry requests for there host affinity is disabled. Call back should return immediately.
      fail("Timed out waiting for the containers to start");
    }

    cpm.onStreamProcessorLaunchSuccess(container);

    // Mock 4th failure not exceeding retry window.
    cpm.getProcessorFailures().put(processorId, new ProcessorFailure(3, Instant.now(), Duration.ZERO));
    cpm.onResourceCompleted(new SamzaResourceStatus(container.getContainerId(), "diagnostics", 1));
    assertEquals(failAfterRetries, cpm.getJobFailureCriteriaMet()); // expecting failed container
    assertEquals(3, cpm.getProcessorFailures().get(processorId).getCount()); // count won't update on failure
    if (failAfterRetries) {
      assertTrue(cpm.shouldShutdown());
    } else {
      assertFalse(cpm.shouldShutdown());
    }
    assertEquals(0, allocator.getContainerRequestState().numPendingRequests());
    assertEquals(0, allocator.getContainerRequestState().numDelayedRequests());

    cpm.stop();
  }
Test scenario where a container fails multiple times but failures are more than retryWindow apart with host affinity
@throws Exception
samza-core/src/test/java/org/apache/samza/clustermanager/TestContainerProcessManager.java