private CompletableFuture<Void> putBlobAsync(MessageInfo messageInfo, ByteBuffer messageBuf, long size) {
    try {
      if (shouldUpload(messageInfo)) {
        CloudBlobMetadata blobMetadata;
        BlobId blobId = (BlobId) messageInfo.getStoreKey();
        boolean isRouterEncrypted;
        isRouterEncrypted = isRouterEncrypted(blobId);
        EncryptionOrigin encryptionOrigin = isRouterEncrypted ? EncryptionOrigin.ROUTER : EncryptionOrigin.NONE;
        boolean encryptThisBlob = requireEncryption && !isRouterEncrypted;
        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();
          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
          size = (encryptedSize == -1) ? size : encryptedSize;
        } 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;
          blobMetadata =
              new CloudBlobMetadata(blobId, messageInfo.getOperationTimeMs(), messageInfo.getExpirationTimeInMs(),
                  messageInfo.getSize(), encryptionOrigin, lifeVersion);
        }

        // Upload blob asynchronously
        return uploadAsyncWithRetries(blobId, messageBuf, size, blobMetadata).handle((uploaded, throwable) -> {
          if (throwable != null) {
            throw throwable instanceof CompletionException ? (CompletionException) throwable
                : new CompletionException(throwable);
          }
          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 CompletionException(
                new StoreException(String.format("Another blob with same key %s exists in store", blobId.getID()),
                    StoreErrorCodes.AlreadyExist));
          } else {
            return null;
          }
        });
      }
      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()) {
        return FutureUtils.completedExceptionally(new StoreException(
            String.format("Another blob with same key %s exists in store", messageInfo.getStoreKey().getID()),
            StoreErrorCodes.AlreadyExist));
      }
      return CompletableFuture.completedFuture(null);
    } catch (IOException e) {
      return FutureUtils.completedExceptionally(e);
    }
  }
Upload the blob to the cloud destination asynchronously.
@param messageInfo the {@link MessageInfo} containing blob metadata.
@param messageBuf the bytes to be uploaded.
@param size the number of bytes to upload.
@return a {@link CompletableFuture} that will eventually complete successfully when the messages are written to
        store or will contain the {@link CloudStorageException} if an error occurred
ambry-cloud/src/main/java/com/github/ambry/cloud/CloudBlobStore.java