public CompletableFuture<StoreInfo> getAsync(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions) {
    try {
      checkStarted();
      checkStoreKeyDuplicates(ids);

      List<MessageInfo> messageInfos = new ArrayList<>(ids.size());
      List<BlobId> blobIdList = ids.stream().map(key -> (BlobId) key).collect(Collectors.toList());

      return cloudDestination.getBlobMetadataAsync(blobIdList).handleAsync((cloudBlobMetadataListMap, throwable) -> {
        if (throwable != null) {
          unwrapCompletionExceptionAndThrowStoreException(throwable, CloudStorageException.class,
              this::mapToStoreException);
          return null;
        } else {
          try {
            // Metadata is fetched successfully. Validate the metadata list and return them to next completion stage.
            if (cloudBlobMetadataListMap.size() < blobIdList.size()) {
              // Throw StoreException with ID_Not_Found if cloudBlobMetadataListMap size is less than expected.
              Set<BlobId> missingBlobs = blobIdList.stream()
                  .filter(blobId -> !cloudBlobMetadataListMap.containsKey(blobId))
                  .collect(Collectors.toSet());
              throw new StoreException("Some of the keys were missing in the cloud metadata store: " + missingBlobs,
                  StoreErrorCodes.IDNotFound);
            }

            // Validate cloud meta data, may throw StoreException with ID_Deleted, TTL_Expired and Authorization_Failure
            long currentTimeStamp = System.currentTimeMillis();
            validateCloudMetadata(cloudBlobMetadataListMap, storeGetOptions, currentTimeStamp, ids);
          } catch (StoreException e) {
            throw new CompletionException(e);
          }

          List<CloudMessageReadSet.BlobReadInfo> blobReadInfos = new ArrayList<>(ids.size());
          for (BlobId blobId : blobIdList) {
            CloudBlobMetadata blobMetadata = cloudBlobMetadataListMap.get(blobId.getID());
            // TODO: need to add ttlUpdated to CloudBlobMetadata so we can use it here
            // For now, set ttlUpdated = true for all permanent blobs, so the correct ttl
            // is applied by GetOperation.
            boolean ttlUpdated = blobMetadata.getExpirationTime() == Utils.Infinite_Time;
            boolean deleted = blobMetadata.getDeletionTime() != Utils.Infinite_Time;
            MessageInfo messageInfo =
                new MessageInfo.Builder(blobId, blobMetadata.getSize(), (short) blobMetadata.getAccountId(),
                    (short) blobMetadata.getContainerId(), getOperationTime(blobMetadata)).isDeleted(deleted)
                    .isTtlUpdated(ttlUpdated)
                    .isUndeleted(blobMetadata.isUndeleted())
                    .expirationTimeInMs(blobMetadata.getExpirationTime())
                    .lifeVersion(blobMetadata.getLifeVersion())
                    .build();
            messageInfos.add(messageInfo);
            blobReadInfos.add(new CloudMessageReadSet.BlobReadInfo(blobMetadata, blobId));
          }
          return blobReadInfos;
        }
      }).thenCompose(blobReadInfos -> {
        // Since prefetch is enabled by default, it should be okay to download the content here. If we want to stream
        // content directly from Azure to output channel, it requires more refactoring of the cloud stack such as
        // passing the output channel to the CloudDestination, StorageClient, etc. We shouldn't need streaming now since
        // each blob stored in Azure is of max 4 MB (as large blobs are chunked to size of 4 MB at router).
        List<CompletableFuture<Void>> operationFutures = new ArrayList<>();
        blobReadInfos.forEach(blobReadInfo -> operationFutures.add(blobReadInfo.downloadBlobAsync(this)));
        return CompletableFuture.allOf(operationFutures.toArray(new CompletableFuture<?>[0])).thenApply(unused -> {
          CloudMessageReadSet messageReadSet = new CloudMessageReadSet(blobReadInfos, this);
          return new StoreInfo(messageReadSet, messageInfos);
        });
      });
    } catch (StoreException e) {
      return FutureUtils.completedExceptionally(e);
    }
  }
Returns the store info for the given ids asynchronously.
@param ids The list of ids whose messages need to be retrieved
@param storeGetOptions A set of additional options that the store needs to use while getting the message
@return a {@link CompletableFuture} that will eventually contain the {@link StoreInfo} for the given ids or the
        {@link StoreException} if an error occurred
ambry-cloud/src/main/java/com/github/ambry/cloud/CloudBlobStore.java