[Segment Replication] Fix bug of replica shard's translog not purging…

… on index flush. (#4928) * Fix bug of replica shard's translog not purging on index flush and fixing related unit tests. Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com> * Update change log Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com> * Update change log message Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com> Signed-off-by: Rishikesh1159 <rishireddy1159@gmail.com>
opensearch-project · Oct 28, 2022 · e34de9f · e34de9f
1 parent 6e90080
commit e34de9f
Show file tree

Hide file tree

Showing 3 changed files with 32 additions and 4 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -171,11 +171,14 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Fix 'org.apache.hc.core5.http.ParseException: Invalid protocol version' under JDK 16+ ([#4827](https://github.com/opensearch-project/OpenSearch/pull/4827))
 - [BUG]: flaky test index/80_geo_point/Single point test([#4860](https://github.com/opensearch-project/OpenSearch/pull/4860))
 - Fix bug in SlicedInputStream with zero length ([#4863](https://github.com/opensearch-project/OpenSearch/pull/4863))
+- Fix a bug on handling an invalid array value for point type field ([#4900](https://github.com/opensearch-project/OpenSearch/pull/4900))
+- [Segment Replication] Fix bug of replica shard's translog not purging on index flush when segment replication is enabled ([4928](https://github.com/opensearch-project/OpenSearch/pull/4928))
 - Fix a bug on handling an invalid array value for point type field #4900([#4900](https://github.com/opensearch-project/OpenSearch/pull/4900))
 - [BUG]: Allow decommission to support delay timeout ([#4930](https://github.com/opensearch-project/OpenSearch/pull/4930))
 - Fix failing test: VerifyVersionConstantsIT ([#4946](https://github.com/opensearch-project/OpenSearch/pull/4946))
 - Fixed compression support for h2c protocol ([#4944](https://github.com/opensearch-project/OpenSearch/pull/4944))
 - Fix for failing checkExtraction, checkLicense and checkNotice tasks for windows gradle check ([#4941](https://github.com/opensearch-project/OpenSearch/pull/4941))
+
 ### Security
 - CVE-2022-25857 org.yaml:snakeyaml DOS vulnerability ([#4341](https://github.com/opensearch-project/OpenSearch/pull/4341))
 

diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java
@@ -129,6 +129,7 @@ public synchronized void updateSegments(final SegmentInfos infos, long seqNo) th
         // lower/higher gens are possible from a new primary that was just elected.
         if (incomingGeneration != lastReceivedGen) {
             commitSegmentInfos();
+            translogManager.getDeletionPolicy().setLocalCheckpointOfSafeCommit(seqNo);
             translogManager.rollTranslogGeneration();
         }
         lastReceivedGen = incomingGeneration;

diff --git a/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java
@@ -246,16 +246,36 @@ public void testReplicaReceivesGenIncrease() throws Exception {
             final IndexShard replica = shards.getReplicas().get(0);
             final int numDocs = randomIntBetween(10, 100);
             shards.indexDocs(numDocs);
+            assertEquals(numDocs, primary.translogStats().estimatedNumberOfOperations());
+            assertEquals(numDocs, replica.translogStats().estimatedNumberOfOperations());
+            assertEquals(numDocs, primary.translogStats().getUncommittedOperations());
+            assertEquals(numDocs, replica.translogStats().getUncommittedOperations());
             flushShard(primary, true);
             replicateSegments(primary, shards.getReplicas());
+            assertEquals(0, primary.translogStats().estimatedNumberOfOperations());
+            assertEquals(0, replica.translogStats().estimatedNumberOfOperations());
+            assertEquals(0, primary.translogStats().getUncommittedOperations());
+            assertEquals(0, replica.translogStats().getUncommittedOperations());
 
-            final int totalDocs = numDocs + shards.indexDocs(randomIntBetween(numDocs + 1, numDocs + 10));
-            flushShard(primary);
+            final int additionalDocs = shards.indexDocs(randomIntBetween(numDocs + 1, numDocs + 10));
+
+            final int totalDocs = numDocs + additionalDocs;
+            primary.refresh("test");
+            replicateSegments(primary, shards.getReplicas());
+            assertEquals(additionalDocs, primary.translogStats().estimatedNumberOfOperations());
+            assertEquals(additionalDocs, replica.translogStats().estimatedNumberOfOperations());
+            assertEquals(additionalDocs, primary.translogStats().getUncommittedOperations());
+            assertEquals(additionalDocs, replica.translogStats().getUncommittedOperations());
+            flushShard(primary, true);
             replicateSegments(primary, shards.getReplicas());
 
             assertEqualCommittedSegments(primary, replica);
             assertDocCount(primary, totalDocs);
             assertDocCount(replica, totalDocs);
+            assertEquals(0, primary.translogStats().estimatedNumberOfOperations());
+            assertEquals(0, replica.translogStats().estimatedNumberOfOperations());
+            assertEquals(0, primary.translogStats().getUncommittedOperations());
+            assertEquals(0, replica.translogStats().getUncommittedOperations());
         }
     }
 
@@ -388,16 +408,20 @@ public void testNRTReplicaPromotedAsPrimary() throws Exception {
             // 2. Create ops that are in the replica's xlog, not in the index.
             // index some more into both but don't replicate. replica will have only numDocs searchable, but should have totalDocs
             // persisted.
-            final int totalDocs = numDocs + shards.indexDocs(randomInt(10));
+            final int additonalDocs = shards.indexDocs(randomInt(10));
+            final int totalDocs = numDocs + additonalDocs;
 
             assertDocCounts(oldPrimary, totalDocs, totalDocs);
             for (IndexShard shard : shards.getReplicas()) {
                 assertDocCounts(shard, totalDocs, numDocs);
             }
+            assertEquals(additonalDocs, nextPrimary.translogStats().estimatedNumberOfOperations());
+            assertEquals(additonalDocs, replica.translogStats().estimatedNumberOfOperations());
+            assertEquals(additonalDocs, nextPrimary.translogStats().getUncommittedOperations());
+            assertEquals(additonalDocs, replica.translogStats().getUncommittedOperations());
 
             // promote the replica
             shards.syncGlobalCheckpoint();
-            assertEquals(totalDocs, nextPrimary.translogStats().estimatedNumberOfOperations());
             shards.promoteReplicaToPrimary(nextPrimary);
 
             // close and start the oldPrimary as a replica.