[Segment Replication] Use engine codec and replica shard ReplicationCheckpoint for replication events (#7732) (#7769)

* [Segment Replication] Use engine codec and replica shard ReplicationCheckpoint for replication events



* Spotless fix



* Fix failing unit test



* Address review comments



* Address review comments rename



* Address review comments



* Add unit test for ForceSegmentSync transport handler



* Add unit test for ForceSegmentSync transport failure



---------


(cherry picked from commit adf7e2c)

Signed-off-by: Suraj Singh <surajrider@gmail.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: opensearch-trigger-bot[bot]

server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTarget.java

@@ -57,14 +57,9 @@ public ReplicationCheckpoint getCheckpoint() {
         return this.checkpoint;
     }
 
-    public SegmentReplicationTarget(
-        ReplicationCheckpoint checkpoint,
-        IndexShard indexShard,
-        SegmentReplicationSource source,
-        ReplicationListener listener
-    ) {
+    public SegmentReplicationTarget(IndexShard indexShard, SegmentReplicationSource source, ReplicationListener listener) {
         super("replication_target", indexShard, new ReplicationLuceneIndex(), listener);
-        this.checkpoint = checkpoint;
+        this.checkpoint = indexShard.getLatestReplicationCheckpoint();
         this.source = source;
         this.state = new SegmentReplicationState(
             indexShard.routingEntry(),
@@ -101,7 +96,7 @@ public SegmentReplicationState state() {
     }
 
     public SegmentReplicationTarget retryCopy() {
-        return new SegmentReplicationTarget(checkpoint, indexShard, source, listener);
+        return new SegmentReplicationTarget(indexShard, source, listener);
     }
 
     @Override

server/src/main/java/org/opensearch/indices/replication/SegmentReplicationTargetService.java

@@ -228,7 +228,7 @@ public synchronized void onNewCheckpoint(final ReplicationCheckpoint receivedChe
             }
             final Thread thread = Thread.currentThread();
             if (replicaShard.shouldProcessCheckpoint(receivedCheckpoint)) {
-                startReplication(receivedCheckpoint, replicaShard, new SegmentReplicationListener() {
+                startReplication(replicaShard, new SegmentReplicationListener() {
                     @Override
                     public void onReplicationDone(SegmentReplicationState state) {
                         logger.trace(
@@ -301,17 +301,8 @@ protected void updateLatestReceivedCheckpoint(ReplicationCheckpoint receivedChec
         }
     }
 
-    public SegmentReplicationTarget startReplication(
-        final ReplicationCheckpoint checkpoint,
-        final IndexShard indexShard,
-        final SegmentReplicationListener listener
-    ) {
-        final SegmentReplicationTarget target = new SegmentReplicationTarget(
-            checkpoint,
-            indexShard,
-            sourceFactory.get(indexShard),
-            listener
-        );
+    public SegmentReplicationTarget startReplication(final IndexShard indexShard, final SegmentReplicationListener listener) {
+        final SegmentReplicationTarget target = new SegmentReplicationTarget(indexShard, sourceFactory.get(indexShard), listener);
         startReplication(target);
         return target;
     }
@@ -429,57 +420,49 @@ public void messageReceived(final ForceSyncRequest request, TransportChannel cha
                 channel.sendResponse(TransportResponse.Empty.INSTANCE);
                 return;
             }
-            startReplication(
-                ReplicationCheckpoint.empty(request.getShardId(), indexShard.getDefaultCodecName()),
-                indexShard,
-                new SegmentReplicationTargetService.SegmentReplicationListener() {
-                    @Override
-                    public void onReplicationDone(SegmentReplicationState state) {
-                        logger.trace(
-                            () -> new ParameterizedMessage(
-                                "[shardId {}] [replication id {}] Replication complete to {}, timing data: {}",
-                                indexShard.shardId().getId(),
-                                state.getReplicationId(),
-                                indexShard.getLatestReplicationCheckpoint(),
-                                state.getTimingData()
-                            )
-                        );
-                        try {
-                            // Promote engine type for primary target
-                            if (indexShard.recoveryState().getPrimary() == true) {
-                                indexShard.resetToWriteableEngine();
-                            }
-                            channel.sendResponse(TransportResponse.Empty.INSTANCE);
-                        } catch (InterruptedException | TimeoutException | IOException e) {
-                            throw new RuntimeException(e);
+            startReplication(indexShard, new SegmentReplicationTargetService.SegmentReplicationListener() {
+                @Override
+                public void onReplicationDone(SegmentReplicationState state) {
+                    logger.trace(
+                        () -> new ParameterizedMessage(
+                            "[shardId {}] [replication id {}] Replication complete to {}, timing data: {}",
+                            indexShard.shardId().getId(),
+                            state.getReplicationId(),
+                            indexShard.getLatestReplicationCheckpoint(),
+                            state.getTimingData()
+                        )
+                    );
+                    try {
+                        // Promote engine type for primary target
+                        if (indexShard.recoveryState().getPrimary() == true) {
+                            indexShard.resetToWriteableEngine();
                         }
+                        channel.sendResponse(TransportResponse.Empty.INSTANCE);
+                    } catch (InterruptedException | TimeoutException | IOException e) {
+                        throw new RuntimeException(e);
                     }
+                }
 
-                    @Override
-                    public void onReplicationFailure(
-                        SegmentReplicationState state,
-                        ReplicationFailedException e,
-                        boolean sendShardFailure
-                    ) {
-                        logger.trace(
-                            () -> new ParameterizedMessage(
-                                "[shardId {}] [replication id {}] Replication failed, timing data: {}",
-                                indexShard.shardId().getId(),
-                                state.getReplicationId(),
-                                state.getTimingData()
-                            )
-                        );
-                        if (sendShardFailure == true) {
-                            indexShard.failShard("replication failure", e);
-                        }
-                        try {
-                            channel.sendResponse(e);
-                        } catch (IOException ex) {
-                            throw new RuntimeException(ex);
-                        }
+                @Override
+                public void onReplicationFailure(SegmentReplicationState state, ReplicationFailedException e, boolean sendShardFailure) {
+                    logger.trace(
+                        () -> new ParameterizedMessage(
+                            "[shardId {}] [replication id {}] Replication failed, timing data: {}",
+                            indexShard.shardId().getId(),
+                            state.getReplicationId(),
+                            state.getTimingData()
+                        )
+                    );
+                    if (sendShardFailure == true) {
+                        indexShard.failShard("replication failure", e);
+                    }
+                    try {
+                        channel.sendResponse(e);
+                    } catch (IOException ex) {
+                        throw new RuntimeException(ex);
                     }
                 }
-            );
+            });
         }
     }
 

server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java

@@ -40,6 +40,7 @@
 import org.opensearch.index.store.StoreFileMetadata;
 import org.opensearch.index.translog.SnapshotMatchers;
 import org.opensearch.index.translog.Translog;
+import org.opensearch.indices.IndicesService;
 import org.opensearch.indices.recovery.RecoverySettings;
 import org.opensearch.indices.recovery.RecoveryTarget;
 import org.opensearch.indices.replication.CheckpointInfoResponse;
@@ -294,7 +295,7 @@ public void testPublishCheckpointAfterRelocationHandOff() throws IOException {
     public void testRejectCheckpointOnShardRoutingPrimary() throws IOException {
         IndexShard primaryShard = newStartedShard(true);
         SegmentReplicationTargetService sut;
-        sut = prepareForReplication(primaryShard, null);
+        sut = prepareForReplication(primaryShard, null, mock(TransportService.class), mock(IndicesService.class));
         SegmentReplicationTargetService spy = spy(sut);
 
         // Starting a new shard in PrimaryMode and shard routing primary.
@@ -314,7 +315,7 @@ public void testRejectCheckpointOnShardRoutingPrimary() throws IOException {
         spy.onNewCheckpoint(new ReplicationCheckpoint(primaryShard.shardId(), 0L, 0L, 0L, Codec.getDefault().getName()), spyShard);
 
         // Verify that checkpoint is not processed as shard routing is primary.
-        verify(spy, times(0)).startReplication(any(), any(), any());
+        verify(spy, times(0)).startReplication(any(), any());
         closeShards(primaryShard);
     }
 
@@ -1027,7 +1028,10 @@ private void assertDocCounts(IndexShard indexShard, int expectedPersistedDocCoun
 
     private void resolveCheckpointInfoResponseListener(ActionListener<CheckpointInfoResponse> listener, IndexShard primary) {
         try {
-            final CopyState copyState = new CopyState(ReplicationCheckpoint.empty(primary.shardId, primary.getDefaultCodecName()), primary);
+            final CopyState copyState = new CopyState(
+                ReplicationCheckpoint.empty(primary.shardId, primary.getLatestReplicationCheckpoint().getCodec()),
+                primary
+            );
             listener.onResponse(
                 new CheckpointInfoResponse(copyState.getCheckpoint(), copyState.getMetadataMap(), copyState.getInfosBytes())
             );
@@ -1041,7 +1045,6 @@ private void startReplicationAndAssertCancellation(IndexShard replica, SegmentRe
         throws InterruptedException {
         CountDownLatch latch = new CountDownLatch(1);
         final SegmentReplicationTarget target = targetService.startReplication(
-            ReplicationCheckpoint.empty(replica.shardId, replica.getDefaultCodecName()),
             replica,
             new SegmentReplicationTargetService.SegmentReplicationListener() {
                 @Override