Fix checkpoint handling to prevent segment replication infinite loop (#18636)

* Fix checkpoint handling to prevent segment replication infinite loop

Signed-off-by: Ashish Singh <[email protected]>

* Modify existing test to verify code change behaviour

Signed-off-by: Ashish Singh <[email protected]>

---------

Signed-off-by: Ashish Singh <[email protected]>

Author: ashking94

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

@@ -49,6 +49,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicLong;
@@ -347,7 +348,10 @@ public void onReplicationDone(SegmentReplicationState state) {
 
                         // if we received a checkpoint during the copy event that is ahead of this
                         // try and process it.
-                        processLatestReceivedCheckpoint(replicaShard, thread);
+                        ReplicationCheckpoint latestReceivedCheckpoint = replicator.getPrimaryCheckpoint(replicaShard.shardId());
+                        if (Objects.nonNull(latestReceivedCheckpoint) && latestReceivedCheckpoint.isAheadOf(receivedCheckpoint)) {
+                            processLatestReceivedCheckpoint(replicaShard, thread);
+                        }
                     }
 
                     @Override

server/src/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java

@@ -625,6 +625,7 @@ public void testStartReplicationListenerSuccess() throws InterruptedException {
 
         latch.await(2, TimeUnit.SECONDS);
         verify(spy, (atLeastOnce())).updateVisibleCheckpoint(eq(0L), eq(replicaShard));
+        verify(spy, times(1)).processLatestReceivedCheckpoint(any(), any());
     }
 
     public void testStartReplicationListenerFailure() throws InterruptedException {
@@ -851,4 +852,5 @@ public void testProcessCheckpointOnClusterStateUpdate() {
         spy.clusterChanged(new ClusterChangedEvent("ignored", oldState, newState));
         verify(spy, times(1)).processLatestReceivedCheckpoint(eq(replicaShard), any());
     }
+
 }