@Override
  public void run() {
    RemoteReplicaInfo cloudReplica = null;
    AmbryPartition partition = null;
    serverScanner = serverReplicationManager.getBackupCheckerThread("ambry_backup_integrity_monitor");
    PartitionIdComparator partitionIdComparator = new PartitionIdComparator();
    try {
      /** Select partition P */
      List<PartitionId> partitions = helixClusterManager.getAllPartitionIds(null);
      PartitionId maxPartitionId = partitions.stream().max(partitionIdComparator).get();
      logger.info("[BackupIntegrityMonitor] Total number of partitions = {}, max partition-id = {}",
          partitions.size(), maxPartitionId.getId());
      partition = (AmbryPartition) partitions.stream()
          .filter(p -> p.getId() == currentPartitionId)
          .collect(Collectors.toList())
          .get(0);
      if (currentPartitionId == 0) {
        // We circled back to partition 0 because we either finished all partitions,
        // or there was some error in finding out the last partition scanned. Clear state.
        logger.info("[BackupIntegrityMonitor] Deleting directory {}", replicationConfig.backupCheckerReportDir);
        Utils.deleteFileOrDirectory(new File(replicationConfig.backupCheckerReportDir));
        logger.info("[BackupIntegrityMonitor] Creating directory {}", replicationConfig.backupCheckerReportDir);
        Files.createDirectories(Paths.get(replicationConfig.backupCheckerReportDir));
      }
      logger.info("[BackupIntegrityMonitor] Verifying backup partition-{}", partition.getId());

      /** Create local Store S */
      BlobStore store = startLocalStore(partition);

      /** Restore cloud backup C */
      logger.info("[BackupIntegrityMonitor] Restoring backup partition-{} to disk [{}]", partition.getId(), store);
      cloudReplica = azureReplicationManager.getCloudReplica(store.getReplicaId());
      // No need to reload tokens, since we clear off disks before each run
      azureReplicator.addRemoteReplicaInfo(cloudReplica);
      RecoveryToken azureToken = (RecoveryToken) cloudReplica.getToken();
      while (!azureToken.isEndOfPartition()) {
        azureReplicator.replicate();
        azureToken = (RecoveryToken) cloudReplica.getToken();
        long numBlobs = azureToken.getNumBlobs();
        if (numBlobs > 0 && (numBlobs % azureConfig.azureBlobStorageMaxResultsPerPage == 0)) {
          // Print progress, if N blobs have been restored from Azure
          logger.info("[BackupIntegrityMonitor] Recovered {} num_azure_blobs {} bytes of partition-{} from Azure Storage",
              numBlobs, azureToken.getBytesRead(), partition.getId());
        }
      }
      logger.info("[BackupIntegrityMonitor] Recovered {} num_azure_blobs {} bytes of partition-{} from Azure Storage",
          azureToken.getNumBlobs(), azureToken.getBytesRead(), partition.getId());

      /** Replicate from server and compare */
      // If we filter for SEALED replicas, then we may return empty as there may be no sealed replicas
      List<AmbryReplica> replicas = partition.getReplicaIds().stream()
          .filter(r -> r.getDataNodeId().getDatacenterName().equals(clusterMapConfig.clustermapVcrDatacenterName))
          .filter(r -> !r.isDown())
          .collect(Collectors.toList());
      if (replicas.isEmpty()) {
        throw new RuntimeException(String.format("[BackupIntegrityMonitor] No server replicas available for partition-%s",
            partition.getId()));
      }

      /** Compare metadata from server replicas with metadata from Azure */
      List<RemoteReplicaInfo> serverReplicas =
          serverReplicationManager.createRemoteReplicaInfos(replicas, store.getReplicaId());
      for (RemoteReplicaInfo serverReplica : serverReplicas) {
        /**
         * Find out how far each replica of the partition has been backed-up in Azure.
         * We will use this information to scan the replica until when it has been backed-up.
         * And then compare data in the replica until that point with the backup in Azure.
         */
        HashMap<String, MessageInfo> azureBlobs = getAzureBlobInfoFromLocalStore(store);
        if (azureToken.getNumBlobs() != azureBlobs.size()) {
          metrics.backupCheckerRuntimeError.inc();
          logger.warn("[BackupIntegrityMonitor] Mismatch, num_azure_blobs = {}, num_azure_blobs on-disk = {}",
              azureToken.getNumBlobs(), azureBlobs.size());
        }
        serverScanner.setAzureBlobInfo(azureBlobs);
        compareMetadata(serverReplica, cloudReplica);
      }

      currentPartitionId = ++currentPartitionId % maxPartitionId.getId();
      // Don't delete any state, leave it for inspection.
    } catch (Throwable e) {
      metrics.backupCheckerRuntimeError.inc();
      logger.error(String.format("[BackupIntegrityMonitor] Failed to verify cloud backup partition-%s due to",
          partition.getId()), e);
      // Swallow all exceptions and print a trace for inspection, but do not kill the job
    }

    try {
      if (cloudReplica != null) {
        azureReplicator.removeRemoteReplicaInfo(cloudReplica);
      }
    } catch (Throwable e) {
      metrics.backupCheckerRuntimeError.inc();
      logger.error("[BackupIntegrityMonitor] Failed to stop due to", e);
    }

    try {
      if (partition != null) {
        stopLocalStore(partition);
      }
    } catch (Throwable e) {
      metrics.backupCheckerRuntimeError.inc();
      logger.error("[BackupIntegrityMonitor] Failed to stop due to", e);
    }
  }
This method randomly picks a partition in a cluster and locates its replica in the cluster and the cloud.
Then it downloads and compares data from both replicas.
ambry-cloud/src/main/java/com/github/ambry/cloud/BackupIntegrityMonitor.java