void handleResponse(ResponseInfo responseInfo, PutResponse putResponse) {
      int correlationId = responseInfo.getRequestInfo().getRequest().getCorrelationId();
      RequestInfo requestInfo = correlationIdToChunkPutRequestInfo.remove(correlationId);
      if (requestInfo == null) {
        // Ignore right away. This could mean:
        // - the response is valid for this chunk, but was timed out and removed from the map.
        // - the response is for an earlier attempt of this chunk (slipped put scenario). And the map was cleared
        // before attempting the slipped put.
        // - the response is for an earlier chunk held by this PutChunk.
        logger.debug("{}: No matching request found for {}", loggingContext, correlationId);
        return;
      }
      if (responseInfo.isQuotaRejected()) {
        processQuotaRejectedResponse(correlationId, requestInfo.getReplicaId());
        return;
      }
      // Track the over all time taken for the response since the creation of the request.
      long requestLatencyMs = time.milliseconds() - requestInfo.getRequestCreateTime();
      routerMetrics.routerRequestLatencyMs.update(requestLatencyMs);
      routerMetrics.getDataNodeBasedMetrics(requestInfo.getReplicaId().getDataNodeId()).putRequestLatencyMs.update(
          requestLatencyMs);
      boolean isSuccessful;
      TrackedRequestFinalState putRequestFinalState = null;
      if (responseInfo.getError() != null) {
        logger.debug("{}: PutRequest with response correlationId {} timed out for replica {} ", loggingContext,
            correlationId, requestInfo.getReplicaId().getDataNodeId());
        setChunkException(new RouterException(
            "Operation timed out because of " + responseInfo.getError() + " at DataNode " + responseInfo.getDataNode(),
            RouterErrorCode.OperationTimedOut));
        isSuccessful = false;
        putRequestFinalState = TrackedRequestFinalState.FAILURE;
      } else {
        if (putResponse == null) {
          logger.debug(
              "{}: PutRequest with response correlationId {} received an unexpected error on response deserialization from replica {} ",
              loggingContext, correlationId, requestInfo.getReplicaId().getDataNodeId());
          setChunkException(new RouterException("Response deserialization received an unexpected error",
              RouterErrorCode.UnexpectedInternalError));
          isSuccessful = false;
          putRequestFinalState = TrackedRequestFinalState.FAILURE;
        } else {
          if (putResponse.getCorrelationId() != correlationId) {
            // The NetworkClient associates a response with a request based on the fact that only one request is sent
            // out over a connection id, and the response received on a connection id must be for the latest request
            // sent over it. The check here ensures that is indeed the case. If not, log an error and fail this request.
            // There is no other way to handle it.
            routerMetrics.unknownReplicaResponseError.inc();
            logger.error(
                "{}: The correlation id in the PutResponse {} is not the same as the correlation id in the associated PutRequest: {}",
                loggingContext, putResponse.getCorrelationId(), correlationId);
            setChunkException(
                new RouterException("Unexpected internal error", RouterErrorCode.UnexpectedInternalError));
            isSuccessful = false;
            putRequestFinalState = TrackedRequestFinalState.FAILURE;
            // we do not notify the ResponseHandler responsible for failure detection as this is an unexpected error.
          } else {
            ServerErrorCode putError = putResponse.getError();
            if (putError == ServerErrorCode.NoError) {
              if (!verifyCRC()) {
                logger.error("{}: PutRequest with response correlationId {}, blob id {} has a mismatch in crc {} ",
                    loggingContext, correlationId, chunkBlobId, requestInfo.getReplicaId().getDataNodeId());
                setChunkException(
                    new RouterException("CRC mismatch of chunk content before and after writing to server",
                        RouterErrorCode.BlobCorrupted));
                isSuccessful = false;
                putRequestFinalState = TrackedRequestFinalState.FAILURE;
              } else {
                logger.trace("{}: The putRequest was successful for chunk {}", loggingContext, chunkIndex);
                isSuccessful = true;
              }
            } else {
              // chunkException will be set within processServerError.
              logger.trace(
                  "{}: Replica {} returned an error {} for a PutRequest with response correlationId : {} and blobId {}",
                  loggingContext, requestInfo.getReplicaId(), putResponse.getError(), putResponse.getCorrelationId(),
                  blobId);
              processServerError(putResponse.getError());
              isSuccessful = false;
              putRequestFinalState =
                  putError == ServerErrorCode.TemporarilyDisabled ? TrackedRequestFinalState.REQUEST_DISABLED
                      : TrackedRequestFinalState.FAILURE;
            }
          }
        }
      }
      if (isSuccessful) {
        operationTracker.onResponse(requestInfo.getReplicaId(), TrackedRequestFinalState.SUCCESS);
        if (RouterUtils.isRemoteReplica(routerConfig, requestInfo.getReplicaId())) {
          logger.trace("{}: Cross colo request successful for remote replica in {} ", loggingContext,
              requestInfo.getReplicaId().getDataNodeId().getDatacenterName());
          routerMetrics.crossColoSuccessCount.inc();

          // We use a separate metric to the latency of remote vs local writes (such requests are routed through
          // ParanoidDurabilityOperationTracker).
          routerMetrics.routerPutRequestRemoteLatencyMs.update(requestLatencyMs);
        } else {
          routerMetrics.routerPutRequestLocalLatencyMs.update(requestLatencyMs);
        }
      } else {
        onErrorResponse(requestInfo.getReplicaId(), putRequestFinalState);
      }
      checkAndMaybeComplete();
    }
This method is the entry point for handling responses received for requests sent out on behalf of this chunk.
For puts, processing involves determining whether the request was successful, notifying the operation
tracker so it can track the status of the operation, and notifying the response handler for failure detection.
Finally, a check is done to determine whether the operation on the chunk is eligible for completion,
if so the operation is completed right away.
@param responseInfo the response received for a request sent out on behalf of this chunk.
@param putResponse the {@link PutResponse} associated with this response.
ambry-router/src/main/java/com/github/ambry/router/PutOperation.java