From 51ae76bced49c223e2f819735c3dbd24372fdc58 Mon Sep 17 00:00:00 2001 From: guojialiang Date: Thu, 5 Feb 2026 16:16:49 +0800 Subject: [PATCH 1/6] reproduce stale ckp exception Signed-off-by: guojialiang --- .../replication/SegmentReplicationIT.java | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java index f2d65677ceec7..b13da462772c3 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java @@ -63,6 +63,8 @@ import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.common.util.set.Sets; +import org.opensearch.core.common.breaker.CircuitBreaker; +import org.opensearch.core.common.breaker.CircuitBreakingException; import org.opensearch.core.common.io.stream.NamedWriteableRegistry; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; import org.opensearch.core.index.shard.ShardId; @@ -77,6 +79,7 @@ import org.opensearch.index.engine.EngineConfig; import org.opensearch.index.engine.NRTReplicationReaderManager; import org.opensearch.index.shard.IndexShard; +import org.opensearch.index.store.StoreFileMetadata; import org.opensearch.indices.recovery.FileChunkRequest; import org.opensearch.indices.replication.checkpoint.PublishCheckpointAction; import org.opensearch.indices.replication.common.ReplicationType; @@ -140,6 +143,81 @@ private static String indexOrAlias() { return randomBoolean() ? INDEX_NAME : "alias"; } + public void testSegmentReplicationWithException() throws Exception { + final String primaryNode = internalCluster().startDataOnlyNode(); + createIndex(INDEX_NAME); + ensureYellowAndNoInitializingShards(INDEX_NAME); + final String replicaNode = internalCluster().startDataOnlyNode(); + ensureGreen(INDEX_NAME); + + MockTransportService primaryTransportService = ((MockTransportService) internalCluster().getInstance( + TransportService.class, + primaryNode + )); + + AtomicBoolean mockException = new AtomicBoolean(true); + CountDownLatch latch1 = new CountDownLatch(1); + CountDownLatch latch2 = new CountDownLatch(1); + + primaryTransportService.addRequestHandlingBehavior( + SegmentReplicationSourceService.Actions.GET_SEGMENT_FILES, + (handler, request, channel, task) -> { + logger.info( + "replicationId {}, get segment files {}", + ((GetSegmentFilesRequest) request).getReplicationId(), + ((GetSegmentFilesRequest) request).getFilesToFetch().stream().map(StoreFileMetadata::name).collect(Collectors.toList()) + ); + if (mockException.get()) { + mockException.set(false); + latch1.countDown(); + latch2.await(); + throw new CircuitBreakingException("mock circuit break exception", CircuitBreaker.Durability.TRANSIENT); + } else { + handler.messageReceived(request, channel, task); + } + } + ); + + // generate _0.si + client().prepareIndex(INDEX_NAME) + .setId(String.valueOf(1)) + .setSource("foo", "bar") + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .get(); + + latch1.await(); + + MockTransportService replicaTransportService = ((MockTransportService) internalCluster().getInstance( + TransportService.class, + replicaNode + )); + replicaTransportService.addRequestHandlingBehavior( + PublishCheckpointAction.ACTION_NAME + TransportReplicationAction.REPLICA_ACTION_SUFFIX, + (handler, request, channel, task) -> { + logger.info("replica receive publish checkpoint request"); + latch2.countDown(); + handler.messageReceived(request, channel, task); + } + ); + + // generate _1.si + client().prepareIndex(INDEX_NAME) + .setId(String.valueOf(2)) + .setSource("foo2", "bar2") + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .get(); + + waitForSearchableDocs(2, primaryNode, replicaNode); + + client().prepareIndex(INDEX_NAME) + .setId(String.valueOf(3)) + .setSource("foo3", "bar3") + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .get(); + + waitForSearchableDocs(3, primaryNode, replicaNode); + } + public void testAcquireLastIndexCommit() throws Exception { final String primaryNode = internalCluster().startDataOnlyNode(); createIndex(INDEX_NAME); From 0a46006fa3f1886543543bacda921696bf826eef Mon Sep 17 00:00:00 2001 From: guojialiang Date: Thu, 5 Feb 2026 17:51:28 +0800 Subject: [PATCH 2/6] fix stale ckp exception Signed-off-by: guojialiang --- .../AbstractSegmentReplicationTarget.java | 8 ++++- .../MergedSegmentReplicationTarget.java | 2 +- .../replication/SegmentReplicationTarget.java | 12 ++++++- .../SegmentReplicationTargetService.java | 35 +++++++++++++++---- .../replication/SegmentReplicator.java | 6 +++- 5 files changed, 53 insertions(+), 10 deletions(-) diff --git a/server/src/main/java/org/opensearch/indices/replication/AbstractSegmentReplicationTarget.java b/server/src/main/java/org/opensearch/indices/replication/AbstractSegmentReplicationTarget.java index 9ef030ca53397..15a5e74cfe59f 100644 --- a/server/src/main/java/org/opensearch/indices/replication/AbstractSegmentReplicationTarget.java +++ b/server/src/main/java/org/opensearch/indices/replication/AbstractSegmentReplicationTarget.java @@ -47,17 +47,20 @@ public abstract class AbstractSegmentReplicationTarget extends ReplicationTarget protected final SegmentReplicationSource source; protected final SegmentReplicationState state; protected final MultiFileWriter multiFileWriter; + protected final boolean isRetry; public AbstractSegmentReplicationTarget( String name, IndexShard indexShard, ReplicationCheckpoint checkpoint, SegmentReplicationSource source, + boolean isRetry, ReplicationListener listener ) { super(name, indexShard, new ReplicationLuceneIndex(), listener); this.checkpoint = checkpoint; this.source = source; + this.isRetry = isRetry; this.state = new SegmentReplicationState( indexShard.routingEntry(), stateIndex, @@ -168,7 +171,10 @@ public void startReplication(ActionListener listener, BiConsumer new ParameterizedMessage("Replica received new replication checkpoint from primary [{}]", receivedCheckpoint)); // if the shard is in any state if (replicaShard.state().equals(IndexShardState.CLOSED)) { @@ -332,7 +341,7 @@ public synchronized void onNewCheckpoint(final ReplicationCheckpoint receivedChe } final Thread thread = Thread.currentThread(); if (replicaShard.shouldProcessCheckpoint(receivedCheckpoint)) { - startReplication(replicaShard, receivedCheckpoint, new SegmentReplicationListener() { + startReplication(replicaShard, receivedCheckpoint, isRetry, new SegmentReplicationListener() { @Override public void onReplicationDone(SegmentReplicationState state) { logger.debug( @@ -366,7 +375,7 @@ public void onReplicationFailure( if (sendShardFailure == true) { failShard(e, replicaShard); } else { - processLatestReceivedCheckpoint(replicaShard, thread); + processLatestReceivedCheckpoint(replicaShard, thread, true); } } }); @@ -479,8 +488,12 @@ private DiscoveryNode getPrimaryNode(ShardRouting primaryShard) { return clusterService.state().nodes().get(primaryShard.currentNodeId()); } - // visible to tests protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Thread thread) { + return processLatestReceivedCheckpoint(replicaShard, thread, false); + } + + // visible to tests + protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Thread thread, boolean isRetry) { final ReplicationCheckpoint latestPublishedCheckpoint = replicator.getPrimaryCheckpoint(replicaShard.shardId()); if (latestPublishedCheckpoint != null) { logger.trace( @@ -494,7 +507,7 @@ protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Threa // if we retry ensure the shard is not in the process of being closed. // it will be removed from indexService's collection before the shard is actually marked as closed. if (indicesService.getShardOrNull(replicaShard.shardId()) != null) { - onNewCheckpoint(replicator.getPrimaryCheckpoint(replicaShard.shardId()), replicaShard); + onNewCheckpoint(replicator.getPrimaryCheckpoint(replicaShard.shardId()), replicaShard, isRetry); } }; // Checks if we are using same thread and forks if necessary. @@ -513,19 +526,29 @@ protected void updateLatestReceivedCheckpoint(ReplicationCheckpoint receivedChec replicator.updateReplicationCheckpointStats(receivedCheckpoint, replicaShard); } + public SegmentReplicationTarget startReplication( + final IndexShard indexShard, + final ReplicationCheckpoint checkpoint, + final SegmentReplicationListener listener + ) { + return startReplication(indexShard, checkpoint, false, listener); + } + /** * Start a round of replication and sync to at least the given checkpoint. * @param indexShard - {@link IndexShard} replica shard * @param checkpoint - {@link ReplicationCheckpoint} checkpoint to sync to + * @param isRetry - is it a retry after failure * @param listener - {@link ReplicationListener} * @return {@link SegmentReplicationTarget} target event orchestrating the event. */ public SegmentReplicationTarget startReplication( final IndexShard indexShard, final ReplicationCheckpoint checkpoint, + final boolean isRetry, final SegmentReplicationListener listener ) { - return replicator.startReplication(indexShard, checkpoint, sourceFactory.get(indexShard), listener); + return replicator.startReplication(indexShard, checkpoint, sourceFactory.get(indexShard), isRetry, listener); } // pkg-private for integration tests diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java index 6664ad69553b0..af91d382fcb4a 100644 --- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java +++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicator.java @@ -80,6 +80,7 @@ public void startReplication(IndexShard shard) { shard, shard.getLatestReplicationCheckpoint(), sourceFactory.get().get(shard), + false, new SegmentReplicationTargetService.SegmentReplicationListener() { @Override public void onReplicationDone(SegmentReplicationState state) { @@ -105,6 +106,8 @@ void setSourceFactory(SegmentReplicationSourceFactory sourceFactory) { * Start a round of replication and sync to at least the given checkpoint. * @param indexShard - {@link IndexShard} replica shard * @param checkpoint - {@link ReplicationCheckpoint} checkpoint to sync to + * @param source - {@link SegmentReplicationSource} segment replication source + * @param isRetry - is it a retry after failure * @param listener - {@link ReplicationListener} * @return {@link SegmentReplicationTarget} target event orchestrating the event. */ @@ -112,9 +115,10 @@ SegmentReplicationTarget startReplication( final IndexShard indexShard, final ReplicationCheckpoint checkpoint, final SegmentReplicationSource source, + final boolean isRetry, final SegmentReplicationTargetService.SegmentReplicationListener listener ) { - final SegmentReplicationTarget target = new SegmentReplicationTarget(indexShard, checkpoint, source, listener); + final SegmentReplicationTarget target = new SegmentReplicationTarget(indexShard, checkpoint, source, isRetry, listener); startReplication(target, indexShard.getRecoverySettings().activityTimeout()); return target; } From fd4ba88f9c3fcac6d8ea2983ef16a06fc5a3d9f9 Mon Sep 17 00:00:00 2001 From: guojialiang Date: Thu, 5 Feb 2026 20:14:01 +0800 Subject: [PATCH 3/6] fix test Signed-off-by: guojialiang --- .../replication/SegmentReplicationTarget.java | 2 +- .../SegmentReplicationTargetService.java | 15 +++++++++++++++ .../SegmentReplicationTargetServiceTests.java | 9 +++++---- 3 files changed, 21 insertions(+), 5 deletions(-) diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java index 8783f4f84e44b..1e9830bf2a213 100644 --- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java +++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java @@ -138,6 +138,6 @@ protected void finalizeReplication(CheckpointInfoResponse checkpointInfoResponse @Override public SegmentReplicationTarget retryCopy() { - return new SegmentReplicationTarget(indexShard, checkpoint, source, listener); + return new SegmentReplicationTarget(indexShard, checkpoint, source, true, listener); } } diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java index c0e5bb61e03ee..3d50292958c23 100644 --- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java +++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java @@ -288,6 +288,13 @@ public SegmentReplicationTarget get(ShardId shardId) { return replicator.get(shardId); } + /** + * Invoked when a new checkpoint is received from a primary shard. + * It checks if a new checkpoint should be processed or not and starts replication if needed. + * + * @param receivedCheckpoint received checkpoint that is checked for processing + * @param replicaShard replica shard on which checkpoint is received + */ public void onNewCheckpoint(final ReplicationCheckpoint receivedCheckpoint, final IndexShard replicaShard) { onNewCheckpoint(receivedCheckpoint, replicaShard, false); } @@ -488,6 +495,7 @@ private DiscoveryNode getPrimaryNode(ShardRouting primaryShard) { return clusterService.state().nodes().get(primaryShard.currentNodeId()); } + // visible to tests protected boolean processLatestReceivedCheckpoint(IndexShard replicaShard, Thread thread) { return processLatestReceivedCheckpoint(replicaShard, thread, false); } @@ -526,6 +534,13 @@ protected void updateLatestReceivedCheckpoint(ReplicationCheckpoint receivedChec replicator.updateReplicationCheckpointStats(receivedCheckpoint, replicaShard); } + /** + * Start a round of replication and sync to at least the given checkpoint. + * @param indexShard - {@link IndexShard} replica shard + * @param checkpoint - {@link ReplicationCheckpoint} checkpoint to sync to + * @param listener - {@link ReplicationListener} + * @return {@link SegmentReplicationTarget} target event orchestrating the event. + */ public SegmentReplicationTarget startReplication( final IndexShard indexShard, final ReplicationCheckpoint checkpoint, diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java index d499b4d9f45d0..377f999eb74e6 100644 --- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java +++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java @@ -68,6 +68,7 @@ import static org.opensearch.index.seqno.SequenceNumbers.NO_OPS_PERFORMED; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doAnswer; @@ -476,7 +477,7 @@ public void cancel() { // ensure the old target is cancelled. and new iteration kicks off. verify(targetSpy, times(1)).cancel("Cancelling stuck target after new primary"); - verify(serviceSpy, times(1)).startReplication(eq(replicaShard), any(), any()); + verify(serviceSpy, times(1)).startReplication(eq(replicaShard), any(), anyBoolean(), any()); } public void testMergedSegmentReplicating_HigherPrimaryTermReceived() throws IOException { @@ -628,10 +629,10 @@ public void testStartReplicationListenerSuccess() throws InterruptedException { SegmentReplicationTargetService spy = spy(sut); CountDownLatch latch = new CountDownLatch(1); doAnswer(i -> { - ((SegmentReplicationTargetService.SegmentReplicationListener) i.getArgument(2)).onReplicationDone(state); + ((SegmentReplicationTargetService.SegmentReplicationListener) i.getArgument(3)).onReplicationDone(state); latch.countDown(); return null; - }).when(spy).startReplication(any(), any(), any()); + }).when(spy).startReplication(any(), any(), anyBoolean(), any()); doNothing().when(spy).updateVisibleCheckpoint(eq(0L), any()); spy.afterIndexShardStarted(replicaShard); @@ -675,7 +676,7 @@ public void testProcessLatestCheckpointIfCheckpointAhead() { doReturn(mock(SegmentReplicationTarget.class)).when(service).startReplication(any(), any(), any()); service.updateLatestReceivedCheckpoint(aheadCheckpoint, replicaShard); service.processLatestReceivedCheckpoint(replicaShard, null); - verify(service, times(1)).startReplication(eq(replicaShard), eq(aheadCheckpoint), any()); + verify(service, times(1)).startReplication(eq(replicaShard), eq(aheadCheckpoint), anyBoolean(), any()); } public void testOnNewCheckpointInvokedOnClosedShardDoesNothing() throws IOException { From 095718a3d3cc3e9e3818c461a3aabedb0ea391bf Mon Sep 17 00:00:00 2001 From: guojialiang Date: Thu, 5 Feb 2026 21:32:45 +0800 Subject: [PATCH 4/6] fix test Signed-off-by: guojialiang --- .../opensearch/indices/replication/SegmentReplicationIT.java | 2 +- .../indices/replication/SegmentReplicationTarget.java | 2 +- .../replication/SegmentReplicationTargetServiceTests.java | 1 + 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java index b13da462772c3..36a20419fd4b7 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java @@ -195,8 +195,8 @@ public void testSegmentReplicationWithException() throws Exception { PublishCheckpointAction.ACTION_NAME + TransportReplicationAction.REPLICA_ACTION_SUFFIX, (handler, request, channel, task) -> { logger.info("replica receive publish checkpoint request"); - latch2.countDown(); handler.messageReceived(request, channel, task); + latch2.countDown(); } ); diff --git a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java index 1e9830bf2a213..794f93b8d457f 100644 --- a/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java +++ b/server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java @@ -138,6 +138,6 @@ protected void finalizeReplication(CheckpointInfoResponse checkpointInfoResponse @Override public SegmentReplicationTarget retryCopy() { - return new SegmentReplicationTarget(indexShard, checkpoint, source, true, listener); + return new SegmentReplicationTarget(indexShard, checkpoint, source, isRetry, listener); } } diff --git a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java index 377f999eb74e6..2ac9629a7bb9d 100644 --- a/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java +++ b/server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java @@ -410,6 +410,7 @@ public void testShardAlreadyReplicating_HigherPrimaryTermReceived() throws Inter // skip post replication actions so we can assert execution counts. This will continue to process bc replica's pterm is not advanced // post replication. doReturn(true).when(serviceSpy).processLatestReceivedCheckpoint(any(), any()); + doReturn(true).when(serviceSpy).processLatestReceivedCheckpoint(any(), any(), anyBoolean()); // Create a Mockito spy of target to stub response of few method calls. CountDownLatch latch = new CountDownLatch(1); From 50ed2b59779687b81d3f36ad5f10ae7637d55626 Mon Sep 17 00:00:00 2001 From: guojialiang Date: Fri, 6 Feb 2026 17:38:06 +0800 Subject: [PATCH 5/6] fix test Signed-off-by: guojialiang --- .../indices/replication/SegmentReplicationIT.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java index 36a20419fd4b7..5315eb947ef63 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationIT.java @@ -143,7 +143,13 @@ private static String indexOrAlias() { return randomBoolean() ? INDEX_NAME : "alias"; } - public void testSegmentReplicationWithException() throws Exception { + public void testLocalSegmentReplicationWithException() throws Exception { + // this test stubs transport calls specific to node-node replication. + assumeFalse( + "Skipping the test as its not compatible with segment replication with remote store.", + segmentReplicationWithRemoteEnabled() + ); + final String primaryNode = internalCluster().startDataOnlyNode(); createIndex(INDEX_NAME); ensureYellowAndNoInitializingShards(INDEX_NAME); From ab707191759b74e26cdca292c4136d522bea6621 Mon Sep 17 00:00:00 2001 From: guojialiang Date: Fri, 6 Feb 2026 17:56:42 +0800 Subject: [PATCH 6/6] add change log Signed-off-by: guojialiang --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c586856f1a889..60b3547ff548c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Fix flaky test failures in ShardsLimitAllocationDeciderIT ([#20375](https://github.com/opensearch-project/OpenSearch/pull/20375)) - Prevent criteria update for context aware indices ([#20250](https://github.com/opensearch-project/OpenSearch/pull/20250)) - Update EncryptedBlobContainer to adhere limits while listing blobs in specific sort order if wrapped blob container supports ([#20514](https://github.com/opensearch-project/OpenSearch/pull/20514)) +- [segment replication] Fix segment replication infinite retry due to stale metadata checkpoint ([#20551](https://github.com/opensearch-project/OpenSearch/pull/20551)) ### Dependencies - Bump `ch.qos.logback:logback-core` and `ch.qos.logback:logback-classic` from 1.5.24 to 1.5.27 ([#20525](https://github.com/opensearch-project/OpenSearch/pull/20525))