@VisibleForTesting
  void handleExpiredRequest(String processorId, String preferredHost,
      SamzaResourceRequest request, ContainerAllocator allocator, ResourceRequestState resourceRequestState) {
    boolean resourceAvailableOnAnyHost = allocator.hasAllocatedResource(ResourceRequestState.ANY_HOST);


    // Case 1. Container placement actions can be taken in either cases of host affinity being set, in both cases
    // mark the container placement action failed
    if (hasActiveContainerPlacementAction(processorId)) {
      resourceRequestState.cancelResourceRequest(request);
      markContainerPlacementActionFailed(getPlacementActionMetadata(processorId).get(),
          "failed the ContainerPlacement action because request for resources to ClusterManager expired");
      return;
    }

    // Case 2. When host affinity is disabled wait for cluster resource manager return resources
    // TODO: SAMZA-2330: Handle expired request for host affinity disabled case by retying request for getting ANY_HOST
    if (!hostAffinityEnabled) {
      return;
    }

    // Case 2. When host affinity is enabled handle the expired requests
    if (standbyContainerManager.isPresent()) {
      standbyContainerManager.get()
          .handleExpiredResourceRequest(processorId, request,
              Optional.ofNullable(allocator.peekAllocatedResource(ResourceRequestState.ANY_HOST)), allocator,
              resourceRequestState);
    } else if (resourceAvailableOnAnyHost) {
      LOG.info("Request for Processor ID: {} on host: {} has expired. Running on ANY_HOST", processorId, preferredHost);
      allocator.runStreamProcessor(request, ResourceRequestState.ANY_HOST);
    } else {
      LOG.info("Request for Processor ID: {} on host: {} has expired. Requesting additional resources on ANY_HOST.",
          processorId, preferredHost);
      resourceRequestState.cancelResourceRequest(request);
      allocator.requestResource(processorId, ResourceRequestState.ANY_HOST);
    }
  }
Handles an expired resource request for both active and standby containers.

Case 1. If this expired request is due to a container placement action mark the request as failed and return
Case 2: Otherwise for a normal resource request following cases are possible
   Case 2.1  If StandbyContainer is present refer to {@code StandbyContainerManager#handleExpiredResourceRequest}
   Case 2.2: host-affinity is enabled, allocator thread looks for allocated resources on ANY_HOST and issues a
             container start if available, otherwise issue an ANY_HOST request
   Case 2.2: host-affinity is disabled, allocator thread does not handle expired requests, it waits for cluster
             manager to return resources on ANY_HOST

@param processorId logical id of the container
@param preferredHost host on which container is requested to be deployed
@param request pending request for the preferred host
@param allocator allocator for requesting resources
@param resourceRequestState state of request in {@link ContainerAllocator}
samza-core/src/main/java/org/apache/samza/clustermanager/ContainerManager.java