server/src/test/java/org/opensearch/index/shard/SegmentReplicationWithNodeToNodeIndexShardTests.java (1)

718-802: Well-structured test for concurrent translog recovery.

The test adequately exercises the concurrent translog recovery path with 200-300 documents and a batch size of 10, creating 20-30 concurrent batches. The assertions comprehensively verify engine types, document counts, translog statistics, and history consistency.

One observation: the batch size of 10 is significantly smaller than the production default of 50,000. While this is intentional for testing, consider adding a brief comment explaining this choice to help future maintainers understand the test design.

+    // Use a small batch size to ensure multiple concurrent batches are created during recovery
     @TestLogging(reason = "Getting trace logs from IndexShard", value = "org.opensearch.index.shard.IndexShard:TRACE")
     public void testPrimaryPromotionWithConcurrentTranslogRecovery() throws Exception {
         final RecoverySettings recoverySettings = new RecoverySettings(

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

server/src/main/java/org/opensearch/index/shard/IndexShard.java (1)

5281-5336: Concurrent translog recovery runner looks solid; consider loosening InternalTranslogManager coupling and documenting assumptions

The new TranslogRecoveryRunner implementation in resetEngineToGlobalCheckpoint() generally looks correct and addresses earlier issues:

• Snapshots created for each batch are now always closed in finally via IOUtils.closeWhileHandlingException, so there is no longer a leak if any task fails.
• Batch loop bounds use a standard ceil‑division pattern, avoiding the extra empty batch when totalOperations is an exact multiple of batchSize.
• Exceptions from individual futures are aggregated into a single IOException, preserving all suppressed causes while still returning the total recovered op count on success.
• Concurrency is conservatively gated on:
  • recoverySettings.isTranslogConcurrentRecoveryEnable(),
  • indexSettings.isSegRepEnabledOrRemoteNode(),
  • totalOperations > batchSize, and
  • the TRANSLOG_RECOVERY executor queue being empty,
    so you fall back to the well‑tested sequential path when conditions aren’t favorable.

A few follow‑ups worth considering:

1. Avoid hard dependency on InternalTranslogManager where possible.
   The assert translogManager instanceof InternalTranslogManager; and the downcast to call getTranslog().newSnapshot(start, end) tie this path to a specific implementation. If you expect other TranslogManager implementations (plugins, future remote variants), it might be cleaner to:

   • either expose a newSnapshot(long fromSeqNo, long toSeqNo) style API on TranslogManager, or
   • guard the concurrent path with an instanceof and fall back to sequential recovery when the manager is not InternalTranslogManager.
     Right now the assert only fires with assertions enabled; in production we’d get a ClassCastException instead.

2. Clarify assumptions around partitioning by seq_no.
   The batching logic derives [start, end] ranges from engine.getProcessedLocalCheckpoint() and batchSize, and uses snapshot.totalOperations() purely to compute the number of batches. This relies on translog invariants (strictly increasing seq_nos, no concurrent appends because operations are blocked during resetEngineToGlobalCheckpoint) so that the union of your per‑batch snapshots matches the original recovery range. A short comment here explaining that assumption would help future maintainers understand why it’s safe to ignore the passed snapshot other than for totalOperations().

3. TRANSLOG_RECOVERY executor assumption.
   The cast to OpenSearchThreadPoolExecutor and queue emptiness check are fine with today’s FixedExecutorBuilder setup for ThreadPool.Names.TRANSLOG_RECOVERY, but they implicitly assume that name will always map to an OpenSearchThreadPoolExecutor. If that ever changes, a runtime ClassCastException will be thrown. A brief comment noting the expectation (or an instanceof guard that falls back to sequential recovery if the cast fails) would make this more robust.

If you’re comfortable with the current constraints (internal manager only and fixed executor type), this can merge as‑is; otherwise the above refactors are low‑risk and can be done now or in a follow‑up.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

server/src/main/java/org/opensearch/index/shard/IndexShard.java (1)

5279-5332: Make concurrent translog recovery robust to snapshot/submit failures and tighten batch logic

The concurrent TranslogRecoveryRunner is close, but there are two important issues:

1. Snapshot leak on failures in the creation loop

   If either:

   • engine.translogManager().newChangesSnapshot(start, end, false) throws, or
   • threadPool.executor(...).submit(...) throws (e.g., RejectedExecutionException during shutdown),

   the method exits before reaching the second loop that closes snapshots. Any snapshots created in prior iterations of the loop are never closed, leaking file descriptors until shard/engine shutdown.

   To fix this, wrap the batch creation + execution phase in a try/finally that always closes every Translog.Snapshot you created, and explicitly close the snapshot if submit() fails before it’s added to the list.

2. Cleaner last-batch detection

   The last-batch check i == totalOperations / batchSize relies on the same arithmetic as the loop bound but is harder to read and easy to get wrong during future changes. Precomputing the batch count and checking against batches - 1 is clearer and matches the ceil formula used in the loop bound.

A minimally invasive refactor that addresses both points:

-        final TranslogRecoveryRunner translogRunner = (snapshot) -> {
-            Engine engine = newEngineReference.get();
-            assert null != engine;
-            int totalOperations = snapshot.totalOperations();
-            long batchSize = recoverySettings.getTranslogConcurrentRecoveryBatchSize();
-            boolean threadPoolNotBusy = ((OpenSearchThreadPoolExecutor) threadPool.executor(ThreadPool.Names.TRANSLOG_RECOVERY)).getQueue()
-                .isEmpty();
-            if (recoverySettings.isTranslogConcurrentRecoveryEnable()
-                && indexSettings.isSegRepEnabledOrRemoteNode()
-                && totalOperations > batchSize
-                && threadPoolNotBusy) {
-                long localCheckpoint = engine.getProcessedLocalCheckpoint();
-                List<Tuple<Future<Integer>, Translog.Snapshot>> translogSnapshotsFutureList = new ArrayList<>();
-                for (int i = 0; i < (totalOperations + batchSize - 1) / batchSize; i++) {
-                    long start = localCheckpoint + 1 + (long) i * batchSize;
-                    long end = (i == totalOperations / batchSize) ? Long.MAX_VALUE : start + batchSize - 1;
-                    Translog.Snapshot translogSnapshot = engine.translogManager().newChangesSnapshot(start, end, false);
-                    translogSnapshotsFutureList.add(
-                        new Tuple<>(
-                            threadPool.executor(ThreadPool.Names.TRANSLOG_RECOVERY)
-                                .submit(() -> runTranslogRecovery(engine, translogSnapshot, Engine.Operation.Origin.LOCAL_RESET, () -> {
-                                    // TODO: add a dedicate recovery stats for the reset translog
-                                })),
-                            translogSnapshot
-                        )
-                    );
-                }
-                Exception exception = null;
-                int totalRecovered = 0;
-                for (Tuple<Future<Integer>, Translog.Snapshot> translogSnapshotFuture : translogSnapshotsFutureList) {
-                    try {
-                        int recoveredOps = translogSnapshotFuture.v1().get();
-                        totalRecovered += recoveredOps;
-                    } catch (Exception e) {
-                        if (exception == null) {
-                            exception = e;
-                        } else {
-                            exception.addSuppressed(e);
-                        }
-                    } finally {
-                        Translog.Snapshot translogSnapshot = translogSnapshotFuture.v2();
-                        IOUtils.closeWhileHandlingException(translogSnapshot);
-                    }
-                }
-                if (exception != null) {
-                    throw new IOException("generate exception when concurrent translog recovery", exception);
-                }
-                return totalRecovered;
-            } else {
-                return runTranslogRecovery(newEngineReference.get(), snapshot, Engine.Operation.Origin.LOCAL_RESET, () -> {
-                    // TODO: add a dedicate recovery stats for the reset translog
-                });
-            }
-        };
+        final TranslogRecoveryRunner translogRunner = (snapshot) -> {
+            Engine engine = newEngineReference.get();
+            assert engine != null;
+            final int totalOperations = snapshot.totalOperations();
+            final long batchSize = recoverySettings.getTranslogConcurrentRecoveryBatchSize();
+            final boolean threadPoolNotBusy = ((OpenSearchThreadPoolExecutor) threadPool.executor(ThreadPool.Names.TRANSLOG_RECOVERY))
+                .getQueue()
+                .isEmpty();
+
+            if (recoverySettings.isTranslogConcurrentRecoveryEnable()
+                && indexSettings.isSegRepEnabledOrRemoteNode()
+                && totalOperations > batchSize
+                && threadPoolNotBusy) {
+                final long localCheckpoint = engine.getProcessedLocalCheckpoint();
+                final long batches = (totalOperations + batchSize - 1) / batchSize;
+                final List<Tuple<Future<Integer>, Translog.Snapshot>> translogSnapshotsFutureList = new ArrayList<>();
+                Exception exception = null;
+                int totalRecovered = 0;
+
+                try {
+                    for (long i = 0; i < batches; i++) {
+                        final long start = localCheckpoint + 1 + i * batchSize;
+                        final long end = (i == batches - 1) ? Long.MAX_VALUE : start + batchSize - 1;
+                        final Translog.Snapshot translogSnapshot = engine.translogManager().newChangesSnapshot(start, end, false);
+                        try {
+                            Future<Integer> future = threadPool.executor(ThreadPool.Names.TRANSLOG_RECOVERY)
+                                .submit(
+                                    () -> runTranslogRecovery(
+                                        engine,
+                                        translogSnapshot,
+                                        Engine.Operation.Origin.LOCAL_RESET,
+                                        () -> {
+                                            // TODO: add a dedicate recovery stats for the reset translog
+                                        }
+                                    )
+                                );
+                            translogSnapshotsFutureList.add(new Tuple<>(future, translogSnapshot));
+                        } catch (RuntimeException e) {
+                            IOUtils.closeWhileHandlingException(translogSnapshot);
+                            throw e;
+                        }
+                    }
+
+                    for (Tuple<Future<Integer>, Translog.Snapshot> translogSnapshotFuture : translogSnapshotsFutureList) {
+                        try {
+                            int recoveredOps = translogSnapshotFuture.v1().get();
+                            totalRecovered += recoveredOps;
+                        } catch (Exception e) {
+                            if (exception == null) {
+                                exception = e;
+                            } else {
+                                exception.addSuppressed(e);
+                            }
+                        }
+                    }
+
+                    if (exception != null) {
+                        throw new IOException("exception during concurrent translog recovery", exception);
+                    }
+                    return totalRecovered;
+                } finally {
+                    for (Tuple<Future<Integer>, Translog.Snapshot> translogSnapshotFuture : translogSnapshotsFutureList) {
+                        IOUtils.closeWhileHandlingException(translogSnapshotFuture.v2());
+                    }
+                }
+            } else {
+                return runTranslogRecovery(newEngineReference.get(), snapshot, Engine.Operation.Origin.LOCAL_RESET, () -> {
+                    // TODO: add a dedicate recovery stats for the reset translog
+                });
+            }
+        };

This keeps the overall behavior and conditions intact, but ensures snapshots are always closed and makes the batch structure easier to reason about.

server/src/test/java/org/opensearch/index/shard/SegmentReplicationWithNodeToNodeIndexShardTests.java (2)

777-781: Consider verifying that concurrent recovery was actually used.

While the test correctly configures concurrent translog recovery and verifies the end result, it doesn't explicitly verify that the concurrent code path was exercised. Consider adding assertions or logging to confirm that concurrent recovery actually occurred (e.g., checking thread pool activity or recovery metrics).

---

717-800: Consider extracting common logic to reduce duplication.

This test method shares significant logic with testNRTReplicaPromotedAsPrimary (lines 533-599). While the duplication is reasonable for testing different scenarios, consider extracting the common test flow into a helper method that accepts RecoverySettings and MergedSegmentPublisher as parameters. This would improve maintainability if the test logic needs to change in the future.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro