Adding logic and tests for rejecting checkpoints if shard is in Prima…

…ryMode. Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com>
opensearch-project · Aug 10, 2022 · a7dfbb8 · a7dfbb8
1 parent 31db9c7
commit a7dfbb8
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 0 deletions.
diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -1431,6 +1431,14 @@ public final boolean shouldProcessCheckpoint(ReplicationCheckpoint requestCheckp
             logger.trace(() -> new ParameterizedMessage("Ignoring new replication checkpoint - shard is not started {}", state()));
             return false;
         }
+        if (getReplicationTracker().isPrimaryMode()) {
+            logger.trace(
+                () -> new ParameterizedMessage(
+                    "Ignoring new replication checkpoint - shard is in primaryMode and cannot receive any checkpoints."
+                )
+            );
+            return false;
+        }
         ReplicationCheckpoint localCheckpoint = getLatestReplicationCheckpoint();
         if (localCheckpoint.isAheadOf(requestCheckpoint)) {
             logger.trace(

diff --git a/...rc/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java b/...rc/test/java/org/opensearch/indices/replication/SegmentReplicationTargetServiceTests.java
@@ -204,6 +204,23 @@ public void testNewCheckpoint_validationPassesAndReplicationFails() throws IOExc
         closeShard(indexShard, false);
     }
 
+    /**
+     * here we are starting a new shard in PrimaryMode and testing that we don't process a checkpoint on shard when it is in PrimaryMode.
+     */
+    public void testRejectCheckpointOnShardPrimaryMode() throws IOException {
+        SegmentReplicationTargetService spy = spy(sut);
+
+        // Starting a new shard in PrimaryMode.
+        IndexShard primaryShard = newStartedShard(true);
+        IndexShard spyShard = spy(primaryShard);
+        doNothing().when(spy).startReplication(any(), any(), any());
+        spy.onNewCheckpoint(aheadCheckpoint, spyShard);
+
+        // Verify that checkpoint is not processed as shard is in PrimaryMode.
+        verify(spy, times(0)).startReplication(any(), any(), any());
+        closeShards(primaryShard);
+    }
+
     public void testReplicationOnDone() throws IOException {
         SegmentReplicationTargetService spy = spy(sut);
         IndexShard spyShard = spy(indexShard);