Distribute lagging replicas across replica groups to reduce head-of-line blocking (#3259)

* Spread laggers across chunks in continuous replication to reduce head-of-line blocking

Adds an opt-in chunking mode for continuous replication that sorts replicas
within each per-remote-DataNode pass by remote lag (descending) and round-robin
distributes them into chunks of at most replicationMaxPartitionCountPerRequest,
instead of slicing sequentially via Utils.partitionList.

Today's sequential slicing can co-locate multiple ambient laggers in one chunk,
where the chunk's iteration time is gated by whichever member has the most data
to fetch — a head-of-line blocking effect. The spread mode guarantees that when
laggers <= chunk count, at most one lagger lands in any one chunk.

Gated by replication.spread.laggers.across.chunks, default false. When false,
behavior is byte-identical to today (delegates to Utils.partitionList). When
true, sorts via a snapshot-then-stride helper. Snapshot is required because
RemoteReplicaInfo.getRemoteLagFromLocalInBytes() reads non-volatile state
written by other threads; re-reading inside the sort comparator can trip
TimSort's contract check and kill the replication cycle.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* Pass ReplicationConfig to DataNodeTracker; drop two primitive config params

Address review feedback: DataNodeTracker now takes ReplicationConfig directly
(matching ReplicaThread / ReplicationManager / ReplicationEngine /
BackupCheckerThread convention) instead of receiving primitive config-derived
values.

Drops two primitive ctor params: maxActiveGroupSize and spreadLaggersAcrossChunks.
DataNodeTracker reads them from replicationConfig.replicationMaxPartitionCountPerRequest
and replicationConfig.replicationSpreadLaggersAcrossChunks internally.

replicaThrottleDurationMs stays as a primitive: it's lane-specific (inter vs intra
DC), and ReplicaThread already resolves it at construction. Passing the resolved
value keeps lane logic where it belongs (in ReplicaThread) rather than duplicating
the lane branch inside DataNodeTracker.

Net: 8-param ctor → 7-param. Behavior unchanged; same fields read.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Qian1900

ambry-api/src/main/java/com/github/ambry/config/ReplicationConfig.java

@@ -29,6 +29,7 @@ public class ReplicationConfig {
   public static final String REPLICATION_MODEL_ACROSS_DATACENTERS = "replication.model.across.datacenters";
   public static final String REPLICATION_ENABLE_CONTINUOUS_REPLICATION = "replication.enable.continuous.replication";
   public static final String REPLICATION_GROUP_ITERATION_LIMIT = "replication.group.iteration.limit";
+  public static final String REPLICATION_SPREAD_LAGGERS_ACROSS_CHUNKS = "replication.spread.laggers.across.chunks";
   public static final String REPLICATION_STANDBY_WAIT_TIMEOUT_TO_TRIGGER_CROSS_COLO_FETCH_SECONDS =
       "replication.standby.wait.timeout.to.trigger.cross.colo.fetch.seconds";
 
@@ -70,6 +71,16 @@ public class ReplicationConfig {
   @Default("1")
   public final int replicationGroupIterationLimit;
 
+  /**
+   * Continuous replication only — no-op when {@code replication.enable.continuous.replication=false}.
+   * When true, replicas within each per-remote-DataNode chunking pass are sorted by remote lag in bytes
+   * (descending), then round-robin distributed into chunks of at most
+   * {@code replicationMaxPartitionCountPerRequest}. When false, replicas are sliced sequentially.
+   */
+  @Config(REPLICATION_SPREAD_LAGGERS_ACROSS_CHUNKS)
+  @Default("false")
+  public final boolean replicationSpreadLaggersAcrossChunks;
+
   /**
    * The number of replica threads on each server that runs the replication protocol for intra dc replication
    */
@@ -407,6 +418,8 @@ public ReplicationConfig(VerifiableProperties verifiableProperties) {
         verifiableProperties.getBoolean(REPLICATION_ENABLE_CONTINUOUS_REPLICATION, false);
     replicationGroupIterationLimit =
         verifiableProperties.getIntInRange(REPLICATION_GROUP_ITERATION_LIMIT, 1, 1, 10000000);
+    replicationSpreadLaggersAcrossChunks =
+        verifiableProperties.getBoolean(REPLICATION_SPREAD_LAGGERS_ACROSS_CHUNKS, false);
     replicationNumOfIntraDCReplicaThreads =
         verifiableProperties.getInt("replication.no.of.intra.dc.replica.threads", 1);
     replicationNumOfInterDCReplicaThreads =

ambry-replication/src/main/java/com/github/ambry/replication/ReplicaThread.java

@@ -2318,8 +2318,8 @@ private void fillDataNodeTrackers() {
         List<RemoteReplicaInfo> remoteReplicasPerNode = entry.getValue();
 
         DataNodeTracker dataNodeTracker =
-            new DataNodeTracker(remoteHost, remoteReplicasPerNode, maxReplicaCountPerRequest, currentStartGroupId, time,
-                threadThrottleDurationMs);
+            new DataNodeTracker(remoteHost, remoteReplicasPerNode, currentStartGroupId, time, threadThrottleDurationMs,
+                replicationConfig, RemoteReplicaInfo::getRemoteLagFromLocalInBytes);
         logger.trace("Thread name: {} for datanode {} create datanode tracker {}", threadName, remoteHost,
             dataNodeTracker);
 

ambry-replication/src/main/java/com/github/ambry/replication/continuous/DataNodeTracker.java

@@ -14,14 +14,18 @@
 package com.github.ambry.replication.continuous;
 
 import com.github.ambry.clustermap.DataNodeId;
+import com.github.ambry.config.ReplicationConfig;
 import com.github.ambry.replication.RemoteReplicaInfo;
 import com.github.ambry.replication.ReplicaThread;
 import com.github.ambry.utils.Time;
 import com.github.ambry.utils.Utils;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Comparator;
+import java.util.IdentityHashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.function.ToLongFunction;
 import java.util.stream.Collectors;
 
 
@@ -42,22 +46,33 @@ public class DataNodeTracker {
    * All group trackers have consecutive group id.
    * @param dataNodeId remote host for which to create datanode tracker
    * @param remoteReplicas remote replicas for this data node
-   * @param maxActiveGroupSize maximum count of replicas in active groups
    * @param startGroupId group id from which we can start and increment and generate unique group id for each group
    * @param time Ambry time
-   * @param replicaThrottleDurationMs throttle duration for replicas
+   * @param replicaThrottleDurationMs throttle duration for replicas. Lane-specific
+   *                                  (inter vs intra DC); resolved once by the owning {@link ReplicaThread}
+   *                                  and passed in as a primitive so lane logic doesn't leak into this class.
+   * @param replicationConfig replication config. Reads
+   *                          {@link ReplicationConfig#replicationMaxPartitionCountPerRequest} for the
+   *                          per-chunk replica cap and
+   *                          {@link ReplicationConfig#replicationSpreadLaggersAcrossChunks} for whether
+   *                          the chunking pass sorts and round-robins instead of slicing sequentially.
+   * @param lagExtractor function that returns a replica's remote-lag-in-bytes; used as the sort key when
+   *                    {@link ReplicationConfig#replicationSpreadLaggersAcrossChunks} is true. Supplied by the
+   *                    caller so the chunking helper does not need access to package-private state on
+   *                    {@link RemoteReplicaInfo}.
    */
-  public DataNodeTracker(DataNodeId dataNodeId, List<RemoteReplicaInfo> remoteReplicas, int maxActiveGroupSize,
-      int startGroupId, Time time, long replicaThrottleDurationMs) {
+  public DataNodeTracker(DataNodeId dataNodeId, List<RemoteReplicaInfo> remoteReplicas, int startGroupId, Time time,
+      long replicaThrottleDurationMs, ReplicationConfig replicationConfig,
+      ToLongFunction<RemoteReplicaInfo> lagExtractor) {
     this.dataNodeId = dataNodeId;
     this.activeGroupTrackers = new ArrayList<>();
 
     int currentGroupId = startGroupId;
 
-    // for this data node break a larger array of remote replicas to smaller multiple arrays of maxActiveGroupSize
+    // for this data node break a larger array of remote replicas to smaller multiple arrays
     List<List<RemoteReplicaInfo>> remoteReplicaSegregatedList =
-        maxActiveGroupSize > 0 ? Utils.partitionList(remoteReplicas, maxActiveGroupSize)
-            : Collections.singletonList(remoteReplicas);
+        chunkReplicas(remoteReplicas, replicationConfig.replicationMaxPartitionCountPerRequest,
+            replicationConfig.replicationSpreadLaggersAcrossChunks, lagExtractor);
 
     // for each of smaller array of remote replicas create active group trackers with consecutive group ids
     for (List<RemoteReplicaInfo> remoteReplicaList : remoteReplicaSegregatedList) {
@@ -147,4 +162,42 @@ public String toString() {
     return "DataNodeTracker :[" + dataNodeId.toString() + " " + activeGroupTrackers.toString() + " "
         + standByGroupTracker.toString() + "]";
   }
+
+  /**
+   * Splits {@code remoteReplicas} into chunks of at most {@code maxActiveGroupSize}. When {@code spread}
+   * is false, replicas are sliced sequentially via {@link Utils#partitionList}. When true, replicas are
+   * first sorted by {@code lagExtractor} descending, then round-robin distributed across chunks so that
+   * the top laggers land in different chunks. When {@code maxActiveGroupSize <= 0}, returns a single
+   * chunk containing all replicas.
+   *
+   * Package-private for test access; do not call from outside {@link DataNodeTracker}.
+   */
+  static List<List<RemoteReplicaInfo>> chunkReplicas(List<RemoteReplicaInfo> remoteReplicas, int maxActiveGroupSize,
+      boolean spread, ToLongFunction<RemoteReplicaInfo> lagExtractor) {
+    if (maxActiveGroupSize <= 0) {
+      return Collections.singletonList(remoteReplicas);
+    }
+    if (!spread || remoteReplicas.isEmpty()) {
+      return Utils.partitionList(remoteReplicas, maxActiveGroupSize);
+    }
+    List<RemoteReplicaInfo> sorted = new ArrayList<>(remoteReplicas);
+    // Snapshot lag once per replica before sorting. Re-reading inside the comparator is unsafe:
+    // the underlying field is mutated by other threads and can trip TimSort's contract check.
+    // IdentityHashMap because RemoteReplicaInfo overrides equals() without hashCode().
+    Map<RemoteReplicaInfo, Long> lagSnapshot = new IdentityHashMap<>(sorted.size());
+    for (RemoteReplicaInfo r : sorted) {
+      lagSnapshot.put(r, lagExtractor.applyAsLong(r));
+    }
+    sorted.sort((a, b) -> Long.compare(lagSnapshot.get(b), lagSnapshot.get(a)));
+    int chunkCount = (sorted.size() + maxActiveGroupSize - 1) / maxActiveGroupSize;
+    List<List<RemoteReplicaInfo>> chunks = new ArrayList<>(chunkCount);
+    for (int chunk = 0; chunk < chunkCount; chunk++) {
+      List<RemoteReplicaInfo> bucket = new ArrayList<>(maxActiveGroupSize);
+      for (int j = chunk; j < sorted.size(); j += chunkCount) {
+        bucket.add(sorted.get(j));
+      }
+      chunks.add(bucket);
+    }
+    return chunks;
+  }
 }