Introduce concurrent translog recovery to accelerate segment replication primary promotion

CHANGELOG.md

@@ -19,6 +19,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add tracing support for StreamingRestChannel  ([#20361](https://github.com/opensearch-project/OpenSearch/pull/20361))
 - Introduce new libs/netty4 module to share common implementation between netty-based plugins and modules (transport-netty4, transport-reactor-netty4) ([#20447](https://github.com/opensearch-project/OpenSearch/pull/20447))
 - Add validation to make crypto store settings immutable ([#20123](https://github.com/opensearch-project/OpenSearch/pull/20123))
+- Introduce concurrent translog recovery to accelerate segment replication primary promotion ([#20251](https://github.com/opensearch-project/OpenSearch/pull/20251))
 - Update to `almalinux:10` ([#20482](https://github.com/opensearch-project/OpenSearch/pull/20482))
 
 ### Changed

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -346,6 +346,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RecoverySettings.INDICES_RECOVERY_MAX_CONCURRENT_REMOTE_STORE_STREAMS_SETTING,
                 RecoverySettings.INDICES_INTERNAL_REMOTE_UPLOAD_TIMEOUT,
                 RecoverySettings.INDICES_RECOVERY_CHUNK_SIZE_SETTING,
+                RecoverySettings.INDICES_TRANSLOG_CONCURRENT_RECOVERY_ENABLE,
+                RecoverySettings.INDICES_TRANSLOG_CONCURRENT_RECOVERY_BATCH_SIZE,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_REPLICAS_RECOVERIES_SETTING,
                 ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_INCOMING_RECOVERIES_SETTING,

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -103,6 +103,7 @@
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.common.util.concurrent.AsyncIOProcessor;
 import org.opensearch.common.util.concurrent.BufferedAsyncIOProcessor;
+import org.opensearch.common.util.concurrent.FutureUtils;
 import org.opensearch.common.util.concurrent.RunOnce;
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.common.util.io.IOUtils;
@@ -238,8 +239,12 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CompletionService;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorCompletionService;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -395,6 +400,9 @@ Runnable getGlobalCheckpointSyncer() {
     private final Set<MergedSegmentCheckpoint> pendingMergedSegmentCheckpoints = Sets.newConcurrentHashSet();
     private final MergedSegmentTransferTracker mergedSegmentTransferTracker;
 
+    // Used to limit the number of concurrent translog tasks. When the semaphore is exhausted, serial recovery is used.
+    private static final Semaphore translogConcurrentRecoverySemaphore = new Semaphore(1000);
+
     @InternalApi
     public IndexShard(
         final ShardRouting shardRouting,
@@ -5274,14 +5282,87 @@ public void close() throws IOException {
             newEngineReference.set(engineFactory.newReadWriteEngine(newEngineConfig(replicationTracker)));
             onNewEngine(newEngineReference.get());
         }
-        final TranslogRecoveryRunner translogRunner = (snapshot) -> runTranslogRecovery(
-            newEngineReference.get(),
-            snapshot,
-            Engine.Operation.Origin.LOCAL_RESET,
-            () -> {
-                // TODO: add a dedicate recovery stats for the reset translog
+        final TranslogRecoveryRunner translogRunner = (snapshot) -> {
+            long startTime = System.currentTimeMillis();
+            Engine engine = newEngineReference.get();
+            assert null != engine;
+            int translogRecoveryOperations;
+            int totalOperations = snapshot.totalOperations();
+            int batchSize = recoverySettings.getTranslogConcurrentRecoveryBatchSize();
+            long localCheckpoint = engine.getProcessedLocalCheckpoint();
+            final int batches = (totalOperations + batchSize - 1) / batchSize;
+            // When the total totalOperations <= batchSize, there is no need to use concurrent execution.
+            boolean isConcurrentRecovery = recoverySettings.isTranslogConcurrentRecoveryEnable()
+                && indexSettings.isSegRepEnabledOrRemoteNode()
+                && totalOperations > batchSize
+                && translogConcurrentRecoverySemaphore.tryAcquire(batches);
+            if (isConcurrentRecovery) {
+                List<Future<Integer>> translogRecoveryFutureList = new ArrayList<>();
+                try {
+                    // Since the translog does not change at this time, it is safe to re-partition the translog snapshot here.
+                    CompletionService<Integer> completionService = new ExecutorCompletionService<>(
+                        threadPool.executor(ThreadPool.Names.TRANSLOG_RECOVERY)
+                    );
+                    for (int i = 0; i < batches; i++) {
+                        long start = localCheckpoint + 1 + (long) i * batchSize;
+                        long end = (i == batches - 1) ? Long.MAX_VALUE : start + batchSize - 1;
+                        translogRecoveryFutureList.add(completionService.submit(() -> {
+                            try (Translog.Snapshot translogSnapshot = engine.translogManager().newChangesSnapshot(start, end, false)) {
+                                return runTranslogRecovery(engine, translogSnapshot, Engine.Operation.Origin.LOCAL_RESET, () -> {
+                                    // TODO: add a dedicate recovery stats for the reset translog
+                                });
+                            }
+                        }));
+                    }
+                    Exception exception = null;
+                    int totalRecovered = 0;
+                    for (int i = 0; i < batches; i++) {
+                        try {
+                            if (exception != null) {
+                                for (Future<Integer> translogRecoveryFuture : translogRecoveryFutureList) {
+                                    FutureUtils.cancel(translogRecoveryFuture);
+                                    if (false == translogRecoveryFuture.isCancelled()) {
+                                        translogRecoveryFuture.get();
+                                    }
+                                }
+                                break;
+                            }
+                            totalRecovered += completionService.take().get();
+                        } catch (Exception e) {
+                            if (exception == null) {
+                                exception = e;
+                            } else {
+                                exception.addSuppressed(e);
+                            }
+                        }
+                    }
+                    if (exception != null) {
+                        throw new IOException("Failed to concurrent recovery translog", exception);
+                    }
+                    translogRecoveryOperations = totalRecovered;
+                } finally {
+                    translogConcurrentRecoverySemaphore.release(batches);
+                }
+            } else {
+                translogRecoveryOperations = runTranslogRecovery(
+                    newEngineReference.get(),
+                    snapshot,
+                    Engine.Operation.Origin.LOCAL_RESET,
+                    () -> {
+                        // TODO: add a dedicate recovery stats for the reset translog
+                    }
+                );
             }
-        );
+
+            logger.info(
+                "translog recovery complete, isConcurrentRecovery {}, cost {}ms, totalRecovered {}",
+                isConcurrentRecovery,
+                System.currentTimeMillis() - startTime,
+                translogRecoveryOperations
+            );
+
+            return translogRecoveryOperations;
+        };
 
         // When the new engine is created, translogs are synced from remote store onto local. Since remote store is the source
         // of truth for translog, we play all translogs that exists locally. Otherwise, the recoverUpto happens upto global checkpoint.

server/src/main/java/org/opensearch/indices/recovery/RecoverySettings.java

@@ -230,6 +230,26 @@ public class RecoverySettings {
         Property.NodeScope
     );
 
+    public static final Setting<Boolean> INDICES_TRANSLOG_CONCURRENT_RECOVERY_ENABLE = Setting.boolSetting(
+        "indices.translog_concurrent_recovery.enable",
+        false,
+        Property.Dynamic,
+        Property.NodeScope
+    );
+
+    // Limiting the maximum value to 1 million is to avoid excessive memory usage of the bitset in LocalCheckpointTracker during
+    // out-of-order execution of concurrent recovery of translog.
+    // Considering the worst-case scenario, with 1000 concurrent recovery tasks, each task recovering 1 million translogs, the bitset
+    // occupancy is approximately 125MB
+    public static final Setting<Integer> INDICES_TRANSLOG_CONCURRENT_RECOVERY_BATCH_SIZE = Setting.intSetting(
+        "indices.translog_concurrent_recovery.batch_size",
+        500000,
+        10000,
+        1000000,
+        Property.Dynamic,
+        Property.NodeScope
+    );
+
     private volatile ByteSizeValue mergedSegmentWarmerMinSegmentSizeThreshold;
     private volatile ByteSizeValue recoveryMaxBytesPerSec;
     private volatile ByteSizeValue replicationMaxBytesPerSec;
@@ -252,6 +272,9 @@ public class RecoverySettings {
     private volatile TimeValue internalRemoteUploadTimeout;
     private volatile TimeValue mergedSegmentReplicationTimeout;
 
+    private volatile boolean isTranslogConcurrentRecoveryEnable;
+    private volatile int translogConcurrentRecoveryBatchSize;
+
     public RecoverySettings(Settings settings, ClusterSettings clusterSettings) {
         this.retryDelayStateSync = INDICES_RECOVERY_RETRY_DELAY_STATE_SYNC_SETTING.get(settings);
         this.maxConcurrentFileChunks = INDICES_RECOVERY_MAX_CONCURRENT_FILE_CHUNKS_SETTING.get(settings);
@@ -286,6 +309,9 @@ public RecoverySettings(Settings settings, ClusterSettings clusterSettings) {
         this.internalRemoteUploadTimeout = INDICES_INTERNAL_REMOTE_UPLOAD_TIMEOUT.get(settings);
         this.chunkSize = INDICES_RECOVERY_CHUNK_SIZE_SETTING.get(settings);
 
+        this.isTranslogConcurrentRecoveryEnable = INDICES_TRANSLOG_CONCURRENT_RECOVERY_ENABLE.get(settings);
+        this.translogConcurrentRecoveryBatchSize = INDICES_TRANSLOG_CONCURRENT_RECOVERY_BATCH_SIZE.get(settings);
+
         clusterSettings.addSettingsUpdateConsumer(INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING, this::setRecoveryMaxBytesPerSec);
         clusterSettings.addSettingsUpdateConsumer(INDICES_REPLICATION_MAX_BYTES_PER_SEC_SETTING, this::setReplicationMaxBytesPerSec);
         clusterSettings.addSettingsUpdateConsumer(
@@ -324,6 +350,11 @@ public RecoverySettings(Settings settings, ClusterSettings clusterSettings) {
             INDICES_RECOVERY_INTERNAL_ACTION_RETRY_TIMEOUT_SETTING,
             this::setInternalActionRetryTimeout
         );
+        clusterSettings.addSettingsUpdateConsumer(INDICES_TRANSLOG_CONCURRENT_RECOVERY_ENABLE, this::setTranslogConcurrentRecoveryEnable);
+        clusterSettings.addSettingsUpdateConsumer(
+            INDICES_TRANSLOG_CONCURRENT_RECOVERY_BATCH_SIZE,
+            this::setTranslogConcurrentRecoveryBatchSize
+        );
     }
 
     private void setMergedSegmentWarmerMinSegmentSizeThreshold(ByteSizeValue value) {
@@ -493,4 +524,19 @@ public void setIndicesMergedSegmentReplicationWarmerEnabled(boolean mergedSegmen
         this.mergedSegmentReplicationWarmerEnabled = mergedSegmentReplicationWarmerEnabled;
     }
 
+    public boolean isTranslogConcurrentRecoveryEnable() {
+        return isTranslogConcurrentRecoveryEnable;
+    }
+
+    private void setTranslogConcurrentRecoveryEnable(boolean translogConcurrentRecoveryEnable) {
+        isTranslogConcurrentRecoveryEnable = translogConcurrentRecoveryEnable;
+    }
+
+    public int getTranslogConcurrentRecoveryBatchSize() {
+        return translogConcurrentRecoveryBatchSize;
+    }
+
+    private void setTranslogConcurrentRecoveryBatchSize(int translogConcurrentRecoveryBatchSize) {
+        this.translogConcurrentRecoveryBatchSize = translogConcurrentRecoveryBatchSize;
+    }
 }

server/src/main/java/org/opensearch/threadpool/ThreadPool.java

@@ -99,6 +99,7 @@ public class ThreadPool implements ReportingService<ThreadPoolInfo>, Scheduler {
     public static class Names {
         public static final String SAME = "same";
         public static final String GENERIC = "generic";
+        public static final String TRANSLOG_RECOVERY = "translog_recovery";
         @Deprecated
         public static final String LISTENER = "listener";
         public static final String GET = "get";
@@ -179,6 +180,7 @@ public static ThreadPoolType fromType(String type) {
         HashMap<String, ThreadPoolType> map = new HashMap<>();
         map.put(Names.SAME, ThreadPoolType.DIRECT);
         map.put(Names.GENERIC, ThreadPoolType.SCALING);
+        map.put(Names.TRANSLOG_RECOVERY, ThreadPoolType.FIXED);
         map.put(Names.LISTENER, ThreadPoolType.FIXED);
         map.put(Names.GET, ThreadPoolType.FIXED);
         map.put(Names.ANALYZE, ThreadPoolType.FIXED);
@@ -258,6 +260,7 @@ public ThreadPool(
         final int genericThreadPoolMax = boundedBy(4 * allocatedProcessors, 128, 512);
         final int snapshotDeletionPoolMax = boundedBy(4 * allocatedProcessors, 64, 256);
         builders.put(Names.GENERIC, new ScalingExecutorBuilder(Names.GENERIC, 4, genericThreadPoolMax, TimeValue.timeValueSeconds(30)));
+        builders.put(Names.TRANSLOG_RECOVERY, new FixedExecutorBuilder(settings, Names.TRANSLOG_RECOVERY, allocatedProcessors, -1));
         builders.put(Names.WRITE, new FixedExecutorBuilder(settings, Names.WRITE, allocatedProcessors, 10000));
         builders.put(Names.GET, new FixedExecutorBuilder(settings, Names.GET, allocatedProcessors, 1000));
         builders.put(Names.ANALYZE, new FixedExecutorBuilder(settings, Names.ANALYZE, 1, 16));