void compareMetadata(RemoteReplicaInfo serverReplica, RemoteReplicaInfo cloudReplica) {
    long partitionId = serverReplica.getReplicaId().getPartitionId().getId();
    try {
      serverScanner.addRemoteReplicaInfo(serverReplica);
      logger.info("[BackupIntegrityMonitor] Starting scanning peer server replica [{}]", serverReplica);
      StoreFindToken newDiskToken = new StoreFindToken(), oldDiskToken = null;
      /**
       * This loop iterates through the remote peer server replica to retrieve metadata for all blobs from the
       * beginning to the current time. The termination condition relies on the token from the peer, which indicates
       * the progress of the scan. The loop stops if this token remains unchanged between successive calls to the
       * replicate() function. The essential requirement is that our scanning process must outpace the growth rate of
       * the replica. Metadata scans, being lighter than full data scans, ensure faster iteration. To ensure timely
       * termination, we must increase replicationFetchSizeInBytes significantly to 128MB or 256MB, from its
       * default of 4MB. This adjustment allows us to advance through the log faster.
       * TODO: Consider implementing a metric to monitor and prevent infinite loop.
       */
      while (!newDiskToken.equals(oldDiskToken)) {
        serverScanner.replicate();
        oldDiskToken = newDiskToken;
        newDiskToken = (StoreFindToken) serverReplica.getToken();
        logger.info("[BackupIntegrityMonitor] Scanned {} num_server_blobs from peer server replica {}",
            serverScanner.getNumBlobScanned(), serverReplica);
      }
      logger.info("[BackupIntegrityMonitor] Completed scanning {} num_server_blobs from peer server replica {}",
          serverScanner.getNumBlobScanned(), serverReplica);

      serverScanner.printKeysAbsentInServer(serverReplica);
      ReplicationMetrics rmetrics = new ReplicationMetrics(cluster.getMetricRegistry(), Collections.emptyList());
      logger.info("[BackupIntegrityMonitor] Verified cloud backup partition-{} against peer server replica {}"
              + ", num_azure_blobs = {}, num_server_blobs = {}, num_mismatches = {}",
          partitionId, serverReplica, ((RecoveryToken) cloudReplica.getToken()).getNumBlobs(),
          serverScanner.getNumBlobScanned(),
          rmetrics.backupIntegrityError.getCount());
    } catch (Throwable e) {
      metrics.backupCheckerRuntimeError.inc();
      logger.error(String.format("[BackupIntegrityMonitor] Failed to verify server replica %s due to",
          serverReplica), e);
      // Swallow all exceptions and print a trace for inspection, but do not kill the job
    }

    try {
      if (serverReplica != null) {
        serverScanner.removeRemoteReplicaInfo(serverReplica);
      }
    } catch (Throwable e) {
      metrics.backupCheckerRuntimeError.inc();
      logger.error(String.format("[BackupIntegrityMonitor] Failed to dequeue server replica %s due to",
          serverReplica), e);
      // Swallow all exceptions and print a trace for inspection, but do not kill the job
    }
  }
Compares metadata received from servers with metadata received from Azure cloud
@param serverReplica
@param cloudReplica
ambry-cloud/src/main/java/com/github/ambry/cloud/BackupIntegrityMonitor.java