@Override
  public void handlePutRequest(NetworkRequest request) throws InterruptedException {

    if (!(request instanceof LocalRequestResponseChannel.LocalChannelRequest)) {
      throw new IllegalArgumentException("The request must be of LocalChannelRequest type");
    }

    PutRequest receivedRequest;

    // This is a case where handlePutRequest is called when frontends are writing to Azure. In this case, this method
    // is called by request handler threads running within the frontend router itself. So, the request can be directly
    // referenced as java objects without any need for deserialization.
    PutRequest sentRequest =
        (PutRequest) ((LocalRequestResponseChannel.LocalChannelRequest) request).getRequestInfo().getRequest();

    // However, we will create a new PutRequest object to represent the received Put request since the blob content
    // 'buffer' in PutRequest is accessed as 'stream' while writing to Store. Also, crc value for this request
    // would be null since it is only calculated (on the fly) when sending the request to network. It might be okay to
    // use null crc here since the scenario for which we are using crc (i.e. possibility of collisions due to fast
    // replication) as described in this PR https://github.com/linkedin/ambry/pull/549 might not be applicable when
    // frontends are talking to Azure.
    receivedRequest = new PutRequest(sentRequest.getCorrelationId(), sentRequest.getClientId(), sentRequest.getBlobId(),
        sentRequest.getBlobProperties(), sentRequest.getUsermetadata(), sentRequest.getBlobSize(),
        sentRequest.getBlobType(), sentRequest.getBlobEncryptionKey(), new ByteBufInputStream(sentRequest.getBlob()),
        null);

    AtomicReference<PutResponse> response = new AtomicReference<>();
    try {
      ServerErrorCode error =
          validateRequest(receivedRequest.getBlobId().getPartition(), RequestOrResponseType.PutRequest, false);
      if (error != ServerErrorCode.NoError) {
        logger.error("Validating put request failed with error {} for request {}", error, receivedRequest);
        requestResponseChannel.sendResponse(
            new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(), error), request, null);
      } else {
        MessageFormatWriteSet writeSet = getMessageFormatWriteSet(receivedRequest);
        CloudBlobStore cloudBlobStore =
            (CloudBlobStore) storeManager.getStore(receivedRequest.getBlobId().getPartition());
        //TODO: Pass executor to have async completion stages run on configured thread pool instead of letting them
        // run on default ForkJoinPool
        cloudBlobStore.putAsync(writeSet).whenCompleteAsync((unused, throwable) -> {
          if (throwable != null) {
            Exception ex = Utils.extractFutureExceptionCause(throwable);
            if (ex instanceof StoreException) {
              StoreErrorCodes storeErrorCode = ((StoreException) ex).getErrorCode();
              logger.error("Store exception on a put with error code {} for request {}", storeErrorCode,
                  receivedRequest, ex);
              response.set(new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
                  ErrorMapping.getStoreErrorMapping(storeErrorCode)));
            } else {
              logger.error("Unknown exception on a put for request {}", receivedRequest, ex);
              response.set(new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
                  ServerErrorCode.UnknownError));
            }
          } else {
            response.set(new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
                ServerErrorCode.NoError));
          }

          try {
            requestResponseChannel.sendResponse(response.get(), request, null);
          } catch (InterruptedException ie) {
            logger.warn("Interrupted while enqueuing the response", ie);
          }
        });
      }
    } catch (MessageFormatException | IOException e) {
      logger.error("Unknown exception on a put for request {}", receivedRequest, e);
      requestResponseChannel.sendResponse(
          new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
              ServerErrorCode.UnknownError), request, null);
    }
  }
@param storeManager {@link StoreManager} object to get stores for replicas
@param requestResponseChannel the {@link RequestResponseChannel} to receive requests and send responses.
@param clusterMap the {@link ClusterMap} of the cluster
@param nodeId the data node Id.
@param registry the {@link MetricRegistry}
ambry-vcr/src/main/java/com/github/ambry/vcr/AmbryCloudRequests.java