@Override
  public StoreInfo get(List<? extends StoreKey> ids, EnumSet<StoreGetOptions> storeGetOptions) throws StoreException {
    // TODO: Remove the duplicate code by calling getAsync() method and wait on it once we verify that download works correctly.
    checkStarted();
    checkStoreKeyDuplicates(ids);
    List<CloudMessageReadSet.BlobReadInfo> blobReadInfos = new ArrayList<>(ids.size());
    List<MessageInfo> messageInfos = new ArrayList<>(ids.size());
    try {
      List<BlobId> blobIdList = ids.stream().map(key -> (BlobId) key).collect(Collectors.toList());
      Map<String, CloudBlobMetadata> cloudBlobMetadataListMap =
          requestAgent.doWithRetries(() -> cloudDestination.getBlobMetadata(blobIdList), "GetBlobMetadata",
              partitionId.toPathString());
      // Throw StoreException with ID_Not_Found if cloudBlobMetadataListMap size is less than expected.
      if (cloudBlobMetadataListMap.size() < blobIdList.size()) {
        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);
      }
      long currentTimeStamp = System.currentTimeMillis();
      // Validate cloud meta data, may throw StoreException with ID_Deleted, TTL_Expired and Authorization_Failure
      validateCloudMetadata(cloudBlobMetadataListMap, storeGetOptions, currentTimeStamp, ids);
      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(blobId, blobMetadata.getSize(), deleted, ttlUpdated, blobMetadata.isUndeleted(),
                blobMetadata.getExpirationTime(), null, (short) blobMetadata.getAccountId(),
                (short) blobMetadata.getContainerId(), getOperationTime(blobMetadata), blobMetadata.getLifeVersion());
        messageInfos.add(messageInfo);
        blobReadInfos.add(new CloudMessageReadSet.BlobReadInfo(blobMetadata, blobId));
      }
    } catch (CloudStorageException e) {
      if (e.getCause() instanceof StoreException) {
        throw (StoreException) e.getCause();
      } else {
        throw new StoreException(e, StoreErrorCodes.IOError);
      }
    }
    CloudMessageReadSet messageReadSet = new CloudMessageReadSet(blobReadInfos, this);
    return new StoreInfo(messageReadSet, messageInfos);
  }
Find key metadata
@param key The key of which blob to return {@link MessageInfo}.
@return
@throws StoreException
ambry-cloud/src/main/java/com/github/ambry/cloud/CloudBlobStore.java