private BlobStore startLocalStore(AmbryPartition partition) throws Exception {
    /**
     * Create local replica L to store cloud data
     * It will be at least as big as the largest replica of the partition on a server.
     * Cloud replica will usually be smaller than server replica, unless server disk has a problem or is compacted.
     */
    long maxReplicaSize = partition.getReplicaIds().stream()
        .map(r -> r.getCapacityInBytes())
        .max(Long::compare)
        .get();
    logger.info("[BackupIntegrityMonitor] Largest replica for partition {} is {} bytes",
        partition.getId(), maxReplicaSize);
    List<DiskId> disks = staticClusterManager.getDataNodeId(nodeId.getHostname(), nodeId.getPort())
        .getDiskIds().stream()
        .filter(d -> d.getState() == HardwareState.AVAILABLE)
        .collect(Collectors.toList());
    logger.info("[BackupIntegrityMonitor] {} disks can accommodate partition-{}", disks.size(), partition.getId());
    // Pick the next disk to make debugging easier
    DiskId disk = disks.get(++currentDiskId % disks.size());
    logger.info("[BackupIntegrityMonitor] Selected disk at mount path {}", disk.getMountPath());
    // Clear disk to make space, this is simpler instead of deciding which partition to delete.
    // This is why this is thread-unsafe.
    Arrays.stream(new File(disk.getMountPath()).listFiles()).forEach(f -> {
      try {
        Utils.deleteFileOrDirectory(f);
      } catch (Throwable e) {
        metrics.backupCheckerRuntimeError.inc();
        throw new RuntimeException(String.format("[BackupIntegrityMonitor] Failed to delete %s due to %s", f, e));
      }
    });
    // Convert Disk object to AmbryDisk object, static-map object -> helix-map object
    AmbryDisk ambryDisk = new AmbryDisk((Disk) disk, clusterMapConfig);
    AmbryServerReplica localReplica = new AmbryServerReplica(clusterMapConfig, partition, ambryDisk,
        true, maxReplicaSize, ReplicaSealStatus.NOT_SEALED);
    if (!storageManager.addBlobStore(localReplica)) {
      metrics.backupCheckerRuntimeError.inc();
      throw new RuntimeException(String.format("[BackupIntegrityMonitor] Failed to start Store for %s",
          partition.getId()));
    }
    Store store = storageManager.getStore(partition, true);
    logger.info("[BackupIntegrityMonitor] Started Store {}", store.toString());
    return (BlobStore) store;
  }
Shut down the monitor waiting for in progress operations to complete.
ambry-cloud/src/main/java/com/github/ambry/cloud/BackupIntegrityMonitor.java