List<ExchangeMetadataResponse> handleReplicaMetadataResponse(ReplicaMetadataResponse response,
      List<RemoteReplicaInfo> replicas, DataNodeId server, List<StoreKey> storeKeysConversionLog) {
    IntStream.range(0, response.getReplicaMetadataResponseInfoList().size())
        .filter(i -> response.getReplicaMetadataResponseInfoList().get(i).getError() == ServerErrorCode.NoError)
        .forEach(i -> {
          ReplicaMetadataResponseInfo metadata = response.getReplicaMetadataResponseInfoList().get(i);
          RemoteReplicaInfo replica = replicas.get(i);
          String output = getFilePath(replica, BLOB_STATE_MISMATCHES_FILE);
          metadata.getMessageInfoList().stream()
              .map(serverBlob -> {
                numBlobScanned.incrementAndGet();
                return mapBlob(serverBlob, storeKeysConversionLog);
              })
              .filter(serverBlob -> serverBlob.getStoreKey() != null)
              .map(serverBlob -> {
                MessageInfo azureBlob = azureBlobMap.remove(serverBlob.getStoreKey().getID());
                Set<BlobMatchStatus> status = azureBlob == null ? Collections.singleton(BLOB_ABSENT_IN_AZURE)
                    : serverBlob.isEqual(azureBlob);
                // if size mismatch, then re-check.
                // this is the only workaround due to different versions on serialization.
                if (status.contains(BLOB_STATE_SIZE_MISMATCH)) {
                  status = recheck(replica, serverBlob, azureBlob, status);
                }
                return new ImmutableTriple(serverBlob, azureBlob, status);
              })
              .filter(tuple -> {
                Set<BlobMatchStatus> status = (Set<BlobMatchStatus>) tuple.getRight();
                // ignore blobs that are intact in Azure, despite a state mismatch between server and cloud
                return !status.contains(BLOB_INTACT_IN_AZURE);
              })
              .filter(tuple -> {
                Set<BlobMatchStatus> status = (Set<BlobMatchStatus>) tuple.getRight();
                // ignore blobs that match between server and azure
                return !status.contains(BLOB_STATE_MATCH);
              })
              .filter(tuple -> {
                MessageInfo serverBlob = (MessageInfo) tuple.getLeft();
                Set<BlobMatchStatus> status = (Set<BlobMatchStatus>) tuple.getRight();
                // ignore blobs that are deleted or expired on server and absent in azure
                // Blob can be absent in azure for 3 reasons:
                // 1. backups haven't caught up
                // 2. cloud compaction executed before server compaction
                // 3. replication ignored such blobs and did not upload them to azure
                return !((serverBlob.isDeleted() || serverBlob.isExpired()) && status.contains(BLOB_ABSENT_IN_AZURE));
              })
              .forEach(tuple -> {
                MessageInfo serverBlob = (MessageInfo) tuple.getLeft();
                MessageInfo azureBlob = (MessageInfo) tuple.getMiddle();
                Set<BlobMatchStatus> status = (Set<BlobMatchStatus>) tuple.getRight();
                String msg = String.join(BackupCheckerFileManager.COLUMN_SEPARATOR,
                    status.stream().map(s -> s.name()).collect(Collectors.joining(",")),
                    MessageInfo.toText(serverBlob), MessageInfo.toText(azureBlob), "\n");
                fileManager.appendToFile(output, msg);
                metrics.backupIntegrityError.inc();
              }); // For each message from replica
          // TODO: To compare data, treat all keys as missing, do not mark rgroup as "done" and compare blob-crc.
          advanceToken(replica, new ExchangeMetadataResponse(Collections.emptySet(), metadata.getFindToken(),
              metadata.getRemoteReplicaLagInBytes(), Collections.emptyMap(), time));
    }); // For each replica-response
    return Collections.emptyList();
  }
Checks each blob from server with its counterpart in cloud backup.
There are 4 cases.
server-blob : cloud-blob
absent : absent  => nothing to do
absent : present => print to file, blob may be compacted or missing on server
present: absent  => print to file, blob may not have been backed up or somehow missing in cloud backup
present: present => check blob-state matches, if not then print to file
@param response The {@link ReplicaMetadataResponse}.
@param replicas
@param server The remote {@link DataNodeId}.
@return
ambry-replication/src/main/java/com/github/ambry/replication/BackupCheckerThread.java