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

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

    GetRequest getRequest;
    // This is a case where handleGetRequest is called when frontends are reading from 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.
    getRequest = (GetRequest) ((LocalRequestResponseChannel.LocalChannelRequest) request).getRequestInfo().getRequest();

    AtomicReference<GetResponse> response = new AtomicReference<>();
    if (getRequest.getPartitionInfoList().size() > 1) {
      logger.error("Invalid number of messages in GET request received from Frontend {}", getRequest);
      response.set(
          new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(), ServerErrorCode.BadRequest));
    }
    PartitionRequestInfo partitionRequestInfo = getRequest.getPartitionInfoList().iterator().next();
    ServerErrorCode error =
        validateRequest(partitionRequestInfo.getPartition(), RequestOrResponseType.GetRequest, false);
    if (error != ServerErrorCode.NoError) {
      logger.error("Validating get request failed for partition {} with error {}", partitionRequestInfo.getPartition(),
          error);
      // Send Response
      requestResponseChannel.sendResponse(
          new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(), error), request, null);
    } else {
      CloudBlobStore cloudBlobStore = (CloudBlobStore) storeManager.getStore(partitionRequestInfo.getPartition());
      EnumSet<StoreGetOptions> storeGetOptions = getStoreGetOptions(getRequest);
      cloudBlobStore.getAsync(partitionRequestInfo.getBlobIds(), storeGetOptions)
          .whenCompleteAsync((info, throwable) -> {
            if (throwable != null) {
              Exception ex = Utils.extractFutureExceptionCause(throwable);
              if (ex instanceof StoreException) {
                logger.error("Store exception on a get with error code {} for partition {}",
                    ((StoreException) ex).getErrorCode(), partitionRequestInfo.getPartition(), ex);
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    ErrorMapping.getStoreErrorMapping(((StoreException) ex).getErrorCode())));
              } else if (ex instanceof MessageFormatException) {
                logger.error("Message format exception on a get with error code {} for partitionRequestInfo {}",
                    ((MessageFormatException) ex).getErrorCode(), partitionRequestInfo, ex);
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    ErrorMapping.getMessageFormatErrorMapping(((MessageFormatException) ex).getErrorCode())));
              } else {
                logger.error("Unknown exception for request {}", getRequest, ex);
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    ServerErrorCode.UnknownError));
              }
            } else {
              MessageFormatSend blobsToSend;
              try {
                blobsToSend = new MessageFormatSend(info.getMessageReadSet(), getRequest.getMessageFormatFlag(),
                    messageFormatMetrics, storeKeyFactory);
                PartitionResponseInfo partitionResponseInfo =
                    new PartitionResponseInfo(partitionRequestInfo.getPartition(), info.getMessageReadSetInfo(),
                        blobsToSend.getMessageMetadataList());
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    Collections.singletonList(partitionResponseInfo), blobsToSend, ServerErrorCode.NoError));
              } catch (IOException e) {
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    ServerErrorCode.UnknownError));
              } catch (MessageFormatException ex) {
                response.set(new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(),
                    ErrorMapping.getMessageFormatErrorMapping(ex.getErrorCode())));
              }
            }
            try {
              // Send Response
              requestResponseChannel.sendResponse(response.get(), request, null);
            } catch (InterruptedException ie) {
              logger.warn("Interrupted while enqueuing the response", ie);
            }
          });
    }
  }
@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