private ReplicaMetadataResponse handleReplicaMetadataRequest(ReplicaMetadataRequest request) {
    List<ReplicaMetadataResponseInfo> responseList =
        new ArrayList<>(request.getReplicaMetadataRequestInfoList().size());

    // For each partition
    for (ReplicaMetadataRequestInfo rinfo : request.getReplicaMetadataRequestInfoList()) {
      // Get previous azure continuation token
      RecoveryToken prevToken = (RecoveryToken) rinfo.getToken();
      String containerName = azureBlobLayoutStrategy.getClusterAwareAzureContainerName(
          rinfo.getPartitionId().toPathString());
      logger.trace("For container {}, previous RecoveryToken = {}", containerName, prevToken.toString());
      if (prevToken.getAmbryPartitionId() > -1 && prevToken.getAmbryPartitionId() != rinfo.getPartitionId().getId()) {
        logger.error("For partition {}, expected RecoveryToken.ambryPartitionId to be {} but found {}",
            rinfo.getPartitionId().getId(), rinfo.getPartitionId().getId(), prevToken.getAmbryPartitionId());
        recoveryMetrics.recoveryTokenError.inc();
        continue;
      }
      if (prevToken.getAzureStorageContainerId() != null &&
          !prevToken.getAzureStorageContainerId().equals(containerName)) {
        logger.error("For partition {}, expected RecoveryToken.azureContainerId to be {} but found {}",
            rinfo.getPartitionId().getId(), containerName, prevToken.getAzureStorageContainerId());
        recoveryMetrics.recoveryTokenError.inc();
        continue;
      }

      List<MessageInfo> messageInfoList = new ArrayList<>();
      if (prevToken.isEndOfPartition()) {
        // End of partition, nothing more to recover assuming nothing was written during recovery
        logger.trace("Recovery reached end-of-partition for {}", rinfo.getPartitionId());
        responseList.add(
            new ReplicaMetadataResponseInfo(rinfo.getPartitionId(), rinfo.getReplicaType(),
                prevToken, messageInfoList, 0,
                ReplicaMetadataResponse.getCompatibleResponseVersion(request.getVersionId())));
        continue;
      }

      // List N blobs with metadata from Azure storage from prev token position
      ListBlobsOptions listBlobsOptions = new ListBlobsOptions()
          .setDetails(new BlobListDetails().setRetrieveMetadata(true))
          .setMaxResultsPerPage(azureCloudConfig.azureBlobStorageMaxResultsPerPage);
      PagedResponse<BlobItem> response;
      try {
        response = azureSyncClient.getBlobStoreCached(containerName)
            .listBlobs(listBlobsOptions, null)
            .iterableByPage(prevToken.getToken())
            .iterator()
            .next();
        recoveryMetrics.listBlobsSuccessRate.mark();
        clientCallback.onListBlobs(rinfo);
      } catch (Throwable t) {
        responseList.add(
            new ReplicaMetadataResponseInfo(rinfo.getPartitionId(), rinfo.getReplicaType(), ServerErrorCode.IOError,
                ReplicaMetadataResponse.getCompatibleResponseVersion(request.getVersionId())));
        logger.error("Failed to list blobs for Ambry partition {} in Azure Storage container {} due to {}",
            rinfo.getPartitionId().getId(), containerName, t);
        recoveryMetrics.listBlobsError.inc();
        continue;
      }

      // Extract ambry metadata
      logger.trace("For container {}, number of blobItems from Azure = {}", containerName, response.getValue().size());
      long bytesRead = 0, blobsRead = 0;
      for (BlobItem blobItem: response.getValue()) {
        MessageInfo messageInfo = getMessageInfo(blobItem);
        if (messageInfo != null) {
          messageInfoList.add(messageInfo);
          messageInfoCache.put(messageInfo.getStoreKey(), messageInfo);
          bytesRead += messageInfo.getSize();
          blobsRead += 1;
        }
      }

      // Save metadata objects
      logger.trace("For container {}, number of messageInfo created = {}", containerName, messageInfoList.size());
      responseList.add(
          new ReplicaMetadataResponseInfo(rinfo.getPartitionId(), rinfo.getReplicaType(),
              new RecoveryToken(prevToken, rinfo.getPartitionId().getId(), containerName,
                  response.getContinuationToken(), blobsRead, bytesRead),
              messageInfoList,
              // Lag metric is useless here as we can't find out size of a container using Azure APIs
              0,
              ReplicaMetadataResponse.getCompatibleResponseVersion(request.getVersionId())));
    }

    // return metadata response
    return new ReplicaMetadataResponse(request.getCorrelationId(), request.getClientId(), ServerErrorCode.NoError,
        responseList, ReplicaMetadataResponse.getCompatibleResponseVersion(request.getVersionId()));
  }
Handle ReplicaMetataRequest and return a response.
@param request The {@link ReplicaMetadataRequest} to handle
@return A {@link ReplicaMetadataResponse}.
ambry-cloud/src/main/java/com/github/ambry/cloud/RecoveryNetworkClient.java