Remove unnecessary refresh listeners from NRTReplicationReaderManager.

This change removes RefreshListeners used by InternalEngine to provide waitFor functionality. These listeners were previously registered onto NRT replicas only to be force released on the next refresh cycle without actually refreshing the reader. This change also removes the unnecessary blocking refresh from NRTReaderManager because we no longer have conflicting refresh invocations from scheduledRefresh. Signed-off-by: Marc Handalian <handalm@amazon.com>
opensearch-project · Jul 25, 2023 · 320723a · 320723a
1 parent a3baa68
commit 320723a
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 17 deletions.
diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java
@@ -77,9 +77,10 @@ public NRTReplicationEngine(EngineConfig engineConfig) {
             this.completionStatsCache = new CompletionStatsCache(() -> acquireSearcher("completion_stats"));
             this.readerManager = readerManager;
             this.readerManager.addListener(completionStatsCache);
-            for (ReferenceManager.RefreshListener listener : engineConfig.getExternalRefreshListener()) {
-                this.readerManager.addListener(listener);
-            }
+            // NRT Replicas do not have a concept of Internal vs External reader managers.
+            // We also do not want to wire up refresh listeners for waitFor & pending refresh location.
+            // which are the current external listeners set from IndexShard.
+            // Only wire up the internal listeners.
             for (ReferenceManager.RefreshListener listener : engineConfig.getInternalRefreshListener()) {
                 this.readerManager.addListener(listener);
             }
@@ -322,22 +323,12 @@ public List<Segment> segments(boolean verbose) {
 
     @Override
     public void refresh(String source) throws EngineException {
-        maybeRefresh(source);
+        // Refresh on this engine should only ever happen in the reader after new segments arrive.
     }
 
     @Override
     public boolean maybeRefresh(String source) throws EngineException {
-        ensureOpen();
-        try {
-            return readerManager.maybeRefresh();
-        } catch (IOException e) {
-            try {
-                failEngine("refresh failed source[" + source + "]", e);
-            } catch (Exception inner) {
-                e.addSuppressed(inner);
-            }
-            throw new RefreshFailedEngineException(shardId, e);
-        }
+        return false;
     }
 
     @Override

diff --git a/server/src/main/java/org/opensearch/index/engine/NRTReplicationReaderManager.java b/server/src/main/java/org/opensearch/index/engine/NRTReplicationReaderManager.java
@@ -103,7 +103,7 @@ public void updateSegments(SegmentInfos infos) throws IOException {
         // is always increased.
         infos.updateGeneration(currentInfos);
         currentInfos = infos;
-        maybeRefreshBlocking();
+        maybeRefresh();
     }
 
     public SegmentInfos getSegmentInfos() {

diff --git a/server/src/main/java/org/opensearch/index/shard/IndexShard.java b/server/src/main/java/org/opensearch/index/shard/IndexShard.java
@@ -4423,7 +4423,8 @@ public void addRefreshListener(Translog.Location location, Consumer<Boolean> lis
                 readAllowed = isReadAllowed();
             }
         }
-        if (readAllowed) {
+        // NRT Replicas will not accept refresh listeners.
+        if (readAllowed && isSegmentReplicationAllowed() == false) {
             refreshListeners.addOrNotify(location, listener);
         } else {
             // we're not yet ready fo ready for reads, just ignore refresh cycles

diff --git a/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/SegmentReplicationIndexShardTests.java
@@ -120,6 +120,12 @@ public void testReplicationCheckpointNotNullForSegRep() throws IOException {
         closeShards(indexShard);
     }
 
+    public void testNRTReplicasDoNotAcceptRefreshListeners() throws IOException {
+        final IndexShard indexShard = newStartedShard(false, settings, new NRTReplicationEngineFactory());
+        indexShard.addRefreshListener(mock(Translog.Location.class), Assert::assertFalse);
+        closeShards(indexShard);
+    }
+
     public void testSegmentInfosAndReplicationCheckpointTuple() throws Exception {
         try (ReplicationGroup shards = createGroup(1, settings, new NRTReplicationEngineFactory())) {
             shards.startAll();