void startPrioritizationCycle() {
    try {
      logger.info("Starting partition prioritization run");

      // 1. Get all bootstrapping partitions from StorageManager
      allBootstrappingPartitions = getAllBootstrappingPartitionsForNode();
      // Update metrics for bootstrapping partitions
      replicationPrioritizationMetrics.updateAllBootstrappingPartitions(allBootstrappingPartitions);

      // Process any completed partitions
      processCompletedPartitions();

      Set<PartitionId> partitionIds = new HashSet<>(allBootstrappingPartitions);
      if (partitionIds.isEmpty()) {
        logger.info("Bootstrapping partition list from StorageManager is empty");
      }

      partitionIds.removeAll(currentlyReplicatingPriorityPartitions);

      if (partitionIds.isEmpty()) {
        logger.info("All bootstrapping partitions are already being replicated");
      }

      Map<PartitionId, List<Operation>> disruptionsByPartition = fetchDisruptions(new ArrayList<>(partitionIds));

      if (disruptionsByPartition == null || disruptionsByPartition.isEmpty()) {
        logger.info("No disruptions found for any partitions");
      } else {
        logger.info("Fetched disruptions for {} partitions", disruptionsByPartition.size());
      }
      // 4. Create prioritized partition lists
      prioritizedPartitions = prioritizePartitions(partitionIds, disruptionsByPartition);
      replicationPrioritizationMetrics.updatePrioritizedPartitions(prioritizedPartitions);

      prioritizedPartitions.keySet().forEach(priorityTier -> logger.info("Found {} partitions in {} category", prioritizedPartitions.get(priorityTier).size(), priorityTier));

      // 5. Update replication priorities
      if (shouldResetReplication()) {
        logger.info("No new high-priority partitions identified "
            + "and no existing high priority run, enabling replication for disabled partitions");

        if (disabledReplicationPartitions.isEmpty()) {
          logger.info("No disabled partitions to enable");
        } else {
          logger.info("Found {} disabled partitions to enable", disabledReplicationPartitions.size());
          resetToNormalReplication();
        }
      } else {
        // Add partitions according to priority order
        Set<PartitionId> highPriorityPartitions = new HashSet<>();

        for (PriorityTier priorityTier : PriorityTier.values()) {
          Set<PartitionId> partitions = prioritizedPartitions.get(priorityTier);
          if (!partitions.isEmpty()) {
            int partitionsToAdd = Math.min(minBatchSizeForHighPriorityPartitions - highPriorityPartitions.size(), partitions.size());
            highPriorityPartitions.addAll(partitions.stream().limit(partitionsToAdd).collect(Collectors.toSet()));

            if (highPriorityPartitions.size() >= minBatchSizeForHighPriorityPartitions) {
              break;
            }
          }
        }

        logger.info("Identified {} high-priority partitions for replication", highPriorityPartitions.size());
        updateReplicationSet(highPriorityPartitions);
      }

    } catch (Exception e) {
      logger.error("Error in partition prioritization task", e);
    }
  }
Starts the replication prioritization manager.
1. Looks at existing high priority partitions and checks if they have completed replication.
2. Identifies bootstrapping partitions from local store
3. From these bootstrapping partitions filter partitions which have count <= MIN_ACTIVE_REPLICAS
4. Fetches disruptions
5. Filters for upcoming disruptions within our window
6. Get affected partitions
7. Control replication thread based on the high priority partitions
ambry-prioritization/src/main/java/com/github/ambry/replica/prioritization/ReplicationPrioritizationManager.java