diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java index 74afbe1a2264c..815da21c84aa1 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/replication/SegmentReplicationRelocationIT.java @@ -60,7 +60,7 @@ public void testPrimaryRelocation() throws Exception { createIndex(1); final String replica = internalCluster().startNode(); ensureGreen(INDEX_NAME); - final int initialDocCount = scaledRandomIntBetween(100, 1000); + final int initialDocCount = scaledRandomIntBetween(10, 100); final WriteRequest.RefreshPolicy refreshPolicy = randomFrom(WriteRequest.RefreshPolicy.values()); final List> pendingIndexResponses = new ArrayList<>(); for (int i = 0; i < initialDocCount; i++) { @@ -137,7 +137,7 @@ public void testPrimaryRelocationWithSegRepFailure() throws Exception { createIndex(1); final String replica = internalCluster().startNode(); ensureGreen(INDEX_NAME); - final int initialDocCount = scaledRandomIntBetween(100, 1000); + final int initialDocCount = scaledRandomIntBetween(10, 100); final WriteRequest.RefreshPolicy refreshPolicy = randomFrom(WriteRequest.RefreshPolicy.values()); final List> pendingIndexResponses = new ArrayList<>(); for (int i = 0; i < initialDocCount; i++) { 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 a7658401bc5de..b52bcf1c5bb6d 100644 --- a/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java +++ b/server/src/main/java/org/opensearch/index/engine/NRTReplicationEngine.java @@ -78,9 +78,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); } @@ -355,22 +356,12 @@ public List 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 d0885301c572a..3604adcc0781b 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -4434,7 +4434,8 @@ public void addRefreshListener(Translog.Location location, Consumer 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();