private void putBlob(MessageInfo messageInfo, ByteBuffer messageBuf, long size)
      throws CloudStorageException, IOException, StoreException {
    // TODO: Remove the duplicate code by calling putBlobAsync() method and wait on it once we verify that upload works
    //  correctly.
    if (shouldUpload(messageInfo)) {
      BlobId blobId = (BlobId) messageInfo.getStoreKey();
      boolean isRouterEncrypted = isRouterEncrypted(blobId);
      EncryptionOrigin encryptionOrigin = isRouterEncrypted ? EncryptionOrigin.ROUTER : EncryptionOrigin.NONE;
      boolean encryptThisBlob = requireEncryption && !isRouterEncrypted;
      boolean uploaded;
      if (encryptThisBlob) {
        // Need to encrypt the buffer before upload
        long encryptedSize = -1;
        Timer.Context encryptionTimer = vcrMetrics.blobEncryptionTime.time();
        try {
          messageBuf = cryptoAgent.encrypt(messageBuf);
          encryptedSize = messageBuf.remaining();
        } catch (GeneralSecurityException ex) {
          vcrMetrics.blobEncryptionErrorCount.inc();
        } finally {
          encryptionTimer.stop();
        }
        vcrMetrics.blobEncryptionCount.inc();
        CloudBlobMetadata blobMetadata =
            new CloudBlobMetadata(blobId, messageInfo.getOperationTimeMs(), messageInfo.getExpirationTimeInMs(),
                messageInfo.getSize(), EncryptionOrigin.VCR, cryptoAgent.getEncryptionContext(),
                cryptoAgentFactory.getClass().getName(), encryptedSize, messageInfo.getLifeVersion());
        // If buffer was encrypted, we no longer know its size
        long bufferLen = (encryptedSize == -1) ? size : encryptedSize;
        uploaded = uploadWithRetries(blobId, messageBuf, bufferLen, blobMetadata);
      } else {
        // PutRequest lifeVersion from frontend is -1. Should set to 0. (0 is the starting life version number for any data).
        // Put from replication or recovery should use liferVersion as it's.
        short lifeVersion =
            messageInfo.hasLifeVersion(messageInfo.getLifeVersion()) ? messageInfo.getLifeVersion() : (short) 0;
        CloudBlobMetadata blobMetadata =
            new CloudBlobMetadata(blobId, messageInfo.getOperationTimeMs(), messageInfo.getExpirationTimeInMs(),
                messageInfo.getSize(), encryptionOrigin, lifeVersion);
        uploaded = uploadWithRetries(blobId, messageBuf, size, blobMetadata);
      }
      addToCache(blobId.getID(), (short) 0, BlobState.CREATED);
      if (!uploaded && !isVcr) {
        // If put is coming from frontend, then uploadBlob must be true. Its not acceptable that a blob already exists.
        // If put is coming from vcr, then findMissingKeys might have reported a key to be missing even though the blob
        // was uploaded.
        throw new StoreException(String.format("Another blob with same key %s exists in store", blobId.getID()),
            StoreErrorCodes.AlreadyExist);
      }
    } else {
      vcrMetrics.blobUploadSkippedCount.inc();
      // The only case where its ok to see a put request for a already seen blob is, during replication if the blob is
      // expiring within {@link CloudConfig#vcrMinTtlDays} for vcr to upload.
      if (isVcr && !isExpiringSoon(messageInfo) && !messageInfo.isDeleted()) {
        throw new StoreException(
            String.format("Another blob with same key %s exists in store", messageInfo.getStoreKey().getID()),
            StoreErrorCodes.AlreadyExist);
      }
    }
  }
Upload the blob to the cloud destination.
@param messageInfo the {@link MessageInfo} containing blob metadata.
@param messageBuf the bytes to be uploaded.
@param size the number of bytes to upload.
@throws CloudStorageException if the upload failed.
ambry-cloud/src/main/java/com/github/ambry/cloud/CloudBlobStore.java