void handleExpiredResource(SamzaResourceRequest request, SamzaResource resource, String preferredHost,
      ResourceRequestState resourceRequestState, ContainerAllocator allocator) {
    LOG.info("Allocated resource {} has expired for Processor ID: {} request: {}. Re-requesting resource again",
        resource, request.getProcessorId(), request);
    resourceRequestState.releaseUnstartableContainer(resource, preferredHost);
    resourceRequestState.cancelResourceRequest(request);
    SamzaResourceRequest newResourceRequest = allocator.getResourceRequest(request.getProcessorId(), request.getPreferredHost());
    if (hasActiveContainerPlacementAction(newResourceRequest.getProcessorId())) {
      ContainerPlacementMetadata metadata = getPlacementActionMetadata(request.getProcessorId()).get();
      metadata.recordResourceRequest(newResourceRequest);
    }
    allocator.issueResourceRequest(newResourceRequest);
  }
Handles expired allocated resource by requesting the same resource again and release the expired allocated resource

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