@Override
  public void handlePutRequest(NetworkRequest request) throws IOException, InterruptedException {
    PutRequest receivedRequest;
    if (request instanceof LocalChannelRequest) {
      // 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) ((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);
    } else {
      InputStream is = request.getInputStream();
      DataInputStream dis = is instanceof DataInputStream ? (DataInputStream) is : new DataInputStream(is);
      receivedRequest = PutRequest.readFrom(dis, clusterMap);
    }
    if (serverConfig.serverEnableBlobPartitionJourneyLogs) {
      logger.info("Received Put Blob request for blob {}, Partition {}", receivedRequest.getBlobId(),
          receivedRequest.getBlobId().getPartition());
    }

    long requestQueueTime = SystemTime.getInstance().milliseconds() - request.getStartTimeInMs();
    long totalTimeSpent = requestQueueTime;
    PutResponse response = null;
    long startTime = SystemTime.getInstance().milliseconds();
    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);
        response = new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(), error);
      } else {
        MessageFormatWriteSet writeSet = getMessageFormatWriteSet(receivedRequest);
        Store storeToPut = storeManager.getStore(receivedRequest.getBlobId().getPartition());
        if (serverConfig.serverEnableBlobPartitionJourneyLogs) {
          logger.info("Blob {} with Partition {} will be put to store {}", receivedRequest.getBlobId(),
              receivedRequest.getBlobId().getPartition(), storeToPut);
        }
        storeToPut.put(writeSet);
        response = new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
            ServerErrorCode.NoError);
        metrics.blobSizeInBytes.update(receivedRequest.getBlobSize());
        metrics.blobUserMetadataSizeInBytes.update(receivedRequest.getUsermetadata().limit());
        if (notification != null) {
          notification.onBlobReplicaCreated(currentNode.getHostname(), currentNode.getPort(),
              receivedRequest.getBlobId().getID(), BlobReplicaSourceType.PRIMARY);
        }
      }
    } catch (StoreException e) {
      logger.error("Store exception on a put with error code {} for request {}", e.getErrorCode(), receivedRequest, e);
      if (e.getErrorCode() == StoreErrorCodes.AlreadyExist) {
        metrics.idAlreadyExistError.inc();
      } else if (e.getErrorCode() == StoreErrorCodes.IOError) {
        metrics.storeIOError.inc();
      } else {
        metrics.unExpectedStorePutError.inc();
      }
      response = new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
          ErrorMapping.getStoreErrorMapping(e.getErrorCode()));
    } catch (Exception e) {
      logger.error("Unknown exception on a put for request {}", receivedRequest, e);
      response = new PutResponse(receivedRequest.getCorrelationId(), receivedRequest.getClientId(),
          ServerErrorCode.UnknownError);
    } finally {
      long processingTime = SystemTime.getInstance().milliseconds() - startTime;
      totalTimeSpent += processingTime;
      publicAccessLogger.info("{} {} processingTime {}", receivedRequest, response, processingTime);
      // Update request metrics.
      RequestMetricsUpdater metricsUpdater =
          new RequestMetricsUpdater(requestQueueTime, processingTime, receivedRequest.getBlobSize(), 0, false);
      receivedRequest.accept(metricsUpdater);
    }
    sendPutResponse(requestResponseChannel, response, request, metrics.putBlobResponseQueueTimeInMs,
        metrics.putBlobSendTimeInMs, metrics.putBlobTotalTimeInMs, totalTimeSpent, receivedRequest.getBlobSize(),
        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