@Override
  public void handleGetRequest(NetworkRequest request) throws IOException, InterruptedException {
    GetRequest getRequest;
    if (request instanceof LocalChannelRequest) {
      // 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) ((LocalChannelRequest) request).getRequestInfo().getRequest();
    } else {
      getRequest = GetRequest.readFrom(new DataInputStream(request.getInputStream()), clusterMap);
    }
    Histogram responseQueueTime;
    Histogram responseSendTime;
    Histogram responseTotalTime;
    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    GetResponse response = null;
    long startTime = SystemTime.getInstance().milliseconds();
    try {
      List<Send> messagesToSendList = new ArrayList<>(getRequest.getPartitionInfoList().size());
      List<PartitionResponseInfo> partitionResponseInfoList = new ArrayList<>(getRequest.getPartitionInfoList().size());
      for (PartitionRequestInfo partitionRequestInfo : getRequest.getPartitionInfoList()) {
        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);
          PartitionResponseInfo partitionResponseInfo =
              new PartitionResponseInfo(partitionRequestInfo.getPartition(), error);
          partitionResponseInfoList.add(partitionResponseInfo);
        } else {
          try {
            Store storeToGet = storeManager.getStore(partitionRequestInfo.getPartition());
            EnumSet<StoreGetOptions> storeGetOptions = getStoreGetOptions(getRequest);
            List<StoreKey> convertedStoreKeys = getConvertedStoreKeys(partitionRequestInfo.getBlobIds());
            List<StoreKey> dedupedStoreKeys =
                convertedStoreKeys.size() > 1 ? convertedStoreKeys.stream().distinct().collect(Collectors.toList())
                    : convertedStoreKeys;
            StoreInfo info = storeToGet.get(dedupedStoreKeys, storeGetOptions);
            MessageFormatSend blobsToSend =
                new MessageFormatSend(info.getMessageReadSet(), getRequest.getMessageFormatFlag(), messageFormatMetrics,
                    storeKeyFactory);
            PartitionResponseInfo partitionResponseInfo =
                new PartitionResponseInfo(partitionRequestInfo.getPartition(), info.getMessageReadSetInfo(),
                    blobsToSend.getMessageMetadataList());
            messagesToSendList.add(blobsToSend);
            partitionResponseInfoList.add(partitionResponseInfo);
          } catch (StoreException e) {
            boolean logInErrorLevel = false;
            if (e.getErrorCode() == StoreErrorCodes.IDNotFound) {
              metrics.idNotFoundError.inc();
            } else if (e.getErrorCode() == StoreErrorCodes.TTLExpired) {
              metrics.ttlExpiredError.inc();
            } else if (e.getErrorCode() == StoreErrorCodes.IDDeleted) {
              metrics.idDeletedError.inc();
            } else if (e.getErrorCode() == StoreErrorCodes.AuthorizationFailure) {
              metrics.getAuthorizationFailure.inc();
            } else {
              metrics.unExpectedStoreGetError.inc();
              logInErrorLevel = true;
            }
            if (logInErrorLevel) {
              logger.error("Store exception on a get with error code {} for partition {}", e.getErrorCode(),
                  partitionRequestInfo.getPartition(), e);
            } else {
              logger.trace("Store exception on a get with error code {} for partition {}", e.getErrorCode(),
                  partitionRequestInfo.getPartition(), e);
            }
            PartitionResponseInfo partitionResponseInfo = new PartitionResponseInfo(partitionRequestInfo.getPartition(),
                ErrorMapping.getStoreErrorMapping(e.getErrorCode()));
            partitionResponseInfoList.add(partitionResponseInfo);
          } catch (MessageFormatException e) {
            logger.error("Message format exception on a get with error code {} for partitionRequestInfo {}",
                e.getErrorCode(), partitionRequestInfo, e);
            if (e.getErrorCode() == MessageFormatErrorCodes.DataCorrupt) {
              metrics.dataCorruptError.inc();
            } else if (e.getErrorCode() == MessageFormatErrorCodes.UnknownFormatVersion) {
              metrics.unknownFormatError.inc();
            }
            PartitionResponseInfo partitionResponseInfo = new PartitionResponseInfo(partitionRequestInfo.getPartition(),
                ErrorMapping.getMessageFormatErrorMapping(e.getErrorCode()));
            partitionResponseInfoList.add(partitionResponseInfo);
          }
        }
      }
      CompositeSend compositeSend = new CompositeSend(messagesToSendList);
      response = new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(), partitionResponseInfoList,
          compositeSend, ServerErrorCode.NoError);
    } catch (Exception e) {
      logger.error("Unknown exception for request {}", getRequest, e);
      response = new GetResponse(getRequest.getCorrelationId(), getRequest.getClientId(), ServerErrorCode.UnknownError);
    } finally {
      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
      totalTimeSpent += processingTime;
      publicAccessLogger.info("{} {} processingTime {}", getRequest, response, processingTime);
      long responseSize = response != null ? response.sizeInBytes() : 0;
      // Update request metrics.
      RequestMetricsUpdater metricsUpdater =
          new RequestMetricsUpdater(requestQueueTime, processingTime, 0, responseSize, false);
      getRequest.accept(metricsUpdater);
      responseQueueTime = metricsUpdater.getResponseQueueTimeHistogram();
      responseSendTime = metricsUpdater.getResponseSendTimeHistogram();
      responseTotalTime = metricsUpdater.getRequestTotalTimeHistogram();
    }
    sendGetResponse(requestResponseChannel, response, request, responseQueueTime, responseSendTime, responseTotalTime,
        totalTimeSpent, response.sizeInBytes(), getRequest.getMessageFormatFlag(), metrics);
  }
The main request implementation class. All requests to the server are
handled by this class
ambry-server/src/main/java/com/github/ambry/server/AmbryRequests.java