From 320723adf455f50465f6724c0afa2a8d19df6efb Mon Sep 17 00:00:00 2001
From: Marc Handalian <handalm@amazon.com>
Date: Mon, 24 Jul 2023 16:53:17 -0700
Subject: [PATCH] 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>
---
 .../index/engine/NRTReplicationEngine.java    | 21 ++++++-------------
 .../engine/NRTReplicationReaderManager.java   |  2 +-
 .../opensearch/index/shard/IndexShard.java    |  3 ++-
 .../SegmentReplicationIndexShardTests.java    |  6 ++++++
 4 files changed, 15 insertions(+), 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
index a9f7a2e70884c..b55508b7facd3 100644
--- 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
index 35409437f605a..7b4c93c7235fe 100644
--- 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
index 0d8a54147d65e..4607fc5bbf8ce 100644
--- 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
index d988e34ef18dc..0c68512f93ea6 100644
--- 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();