private void cleanupExpiredInFlightRequests(RequestRegistrationCallback<PutOperation> requestRegistrationCallback) {
      Iterator<Map.Entry<Integer, RequestInfo>> inFlightRequestsIterator =
          correlationIdToChunkPutRequestInfo.entrySet().iterator();
      while (inFlightRequestsIterator.hasNext()) {
        Map.Entry<Integer, RequestInfo> entry = inFlightRequestsIterator.next();
        int correlationId = entry.getKey();
        RequestInfo requestInfo = entry.getValue();
        // If request times out due to no response from server or due to being stuck in router itself (due to bandwidth
        // throttling, etc) for long time, drop the request.
        long currentTimeInMs = time.milliseconds();
        RouterUtils.RouterRequestExpiryReason routerRequestExpiryReason =
            RouterUtils.isRequestExpired(requestInfo, currentTimeInMs);
        if (routerRequestExpiryReason != RouterUtils.RouterRequestExpiryReason.NO_TIMEOUT) {
          onErrorResponse(requestInfo.getReplicaId(), TrackedRequestFinalState.FAILURE);
          logger.warn("{}: PutRequest with correlationId {} in flight has expired for replica {} {} due to {} ",
              loggingContext, correlationId, requestInfo.getReplicaId().getDataNodeId(), requestInfo,
              routerRequestExpiryReason.name());
          // Do not notify this as a failure to the response handler, as this timeout could simply be due to
          // connection unavailability. If there is indeed a network error, the NetworkClient will provide an error
          // response and the response handler will be notified accordingly.
          setChunkException(RouterUtils.buildTimeoutException(correlationId, requestInfo.getReplicaId().getDataNodeId(),
              chunkBlobId));
          requestRegistrationCallback.registerRequestToDrop(correlationId);
          inFlightRequestsIterator.remove();
          RouterUtils.logTimeoutMetrics(routerRequestExpiryReason, routerMetrics, requestInfo);
        } else {
          // Note: Even though the requests are ordered by correlation id and their creation time, we cannot break out of
          // the while loop here. This is because time outs for all requests may not be equal now.

          // For example, request 1 in the map may have been assigned high time out since it might be sent at a
          // time when the load is high and request 2 may have been assigned lower time out value since the load might have
          // decreased by the time it is sent out. In this case, we should continue iterating the loop and clean up
          // request 2 in the map.

          // The cost of iterating all entries should be okay since the map contains outstanding requests whose number
          // should be small. The maximum outstanding requests possible would be equal to the operation parallelism value
          // and may be few more if adaptive operation tracker is used.
        }
      }
    }
Clean up requests sent out by this operation that have now timed out.
@param requestRegistrationCallback The callback to use to notify the networking layer of dropped requests.
ambry-router/src/main/java/com/github/ambry/router/PutOperation.java