From 2e33e89c075c4b68f3bc472091c49a75361465f6 Mon Sep 17 00:00:00 2001 From: panguixin Date: Tue, 26 May 2026 15:14:35 +0800 Subject: [PATCH] Revert non-closing reader wrapper cache This reverts commit 07540c1a5c361e0b29908a5b34d3e5f605ac4a18. The cached NonClosingReaderWrapper can be closed through one wrapped searcher and later reused by another wrapped searcher for the same underlying reader. Reusing the closed wrapper can cause AlreadyClosedException when the next caller tries to access its leaves. Signed-off-by: panguixin --- .../opensearch/index/shard/IndexShard.java | 72 ++----------------- .../index/shard/IndexShardTests.java | 48 ------------- 2 files changed, 6 insertions(+), 114 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 index 7444460d694cc..90bb8f3cb050b 100644 --- a/server/src/main/java/org/opensearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/opensearch/index/shard/IndexShard.java @@ -252,7 +252,6 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.CompletionService; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorCompletionService; @@ -333,8 +332,6 @@ public class IndexShard extends AbstractIndexShardComponent implements IndicesCl private final IndexingOperationListener indexingOperationListeners; private final Runnable globalCheckpointSyncer; - private final ConcurrentHashMap nonClosingReaderWrapperCache = new ConcurrentHashMap<>(); - private final Function nonClosingReaderWrapperSupplier; Runnable getGlobalCheckpointSyncer() { return globalCheckpointSyncer; @@ -559,34 +556,6 @@ public boolean shouldCache(Query query) { } else { readerWrapper = indexReaderWrapper; } - - nonClosingReaderWrapperSupplier = directoryReader -> { - int[] fromCache = new int[] { 0 }; - try { - // To prevent instantiating a new NonClosingReaderWrapper per query/get/update request, - // the wrapper can be shared across all uses of the same NonClosingReaderWrapper. - return nonClosingReaderWrapperCache.computeIfAbsent(directoryReader, key -> { - try { - NonClosingReaderWrapper closingReaderWrapper = new NonClosingReaderWrapper(key); - fromCache[0] = 1; - return closingReaderWrapper; - } catch (IOException e) { - fromCache[0] = 2; - throw new OpenSearchException("failed to wrap searcher", e); - } - }); - } finally { - if (fromCache[0] == 1) { - OpenSearchDirectoryReader.addReaderCloseListener( - directoryReader, - cacheKey -> nonClosingReaderWrapperCache.remove(directoryReader) - ); - } else if (fromCache[0] == 2) { - nonClosingReaderWrapperCache.remove(directoryReader); - } - } - }; - refreshListeners = buildRefreshListeners(); lastSearcherAccess.set(threadPool.relativeTimeInMillis()); persistMetadata(path, indexSettings, shardRouting, null, logger); @@ -2448,9 +2417,7 @@ private Engine.Searcher wrapSearcher(Engine.Searcher searcher) { : "DirectoryReader must be an instance or OpenSearchDirectoryReader"; boolean success = false; try { - final Engine.Searcher newSearcher = readerWrapper == null - ? searcher - : wrapSearcher(searcher, readerWrapper, nonClosingReaderWrapperSupplier); + final Engine.Searcher newSearcher = readerWrapper == null ? searcher : wrapSearcher(searcher, readerWrapper); assert newSearcher != null; success = true; return newSearcher; @@ -2469,29 +2436,15 @@ private Engine.Searcher wrapSearcher(Engine.Searcher searcher) { public static Engine.Searcher wrapSearcher( Engine.Searcher engineSearcher, CheckedFunction readerWrapper - ) throws IOException { - return wrapSearcher(engineSearcher, readerWrapper, null); - } - - public static Engine.Searcher wrapSearcher( - Engine.Searcher engineSearcher, - CheckedFunction readerWrapper, - Function nonClosingReaderWrapperSupplier ) throws IOException { assert readerWrapper != null; - DirectoryReader directoryReader = engineSearcher.getDirectoryReader(); - final OpenSearchDirectoryReader openSearchDirectoryReader = OpenSearchDirectoryReader.getOpenSearchDirectoryReader(directoryReader); + final OpenSearchDirectoryReader openSearchDirectoryReader = OpenSearchDirectoryReader.getOpenSearchDirectoryReader( + engineSearcher.getDirectoryReader() + ); if (openSearchDirectoryReader == null) { throw new IllegalStateException("Can't wrap non opensearch directory reader"); } - - DirectoryReader nonClosingReaderWrapper; - if (nonClosingReaderWrapperSupplier == null) { - nonClosingReaderWrapper = new NonClosingReaderWrapper(directoryReader); - } else { - nonClosingReaderWrapper = nonClosingReaderWrapperSupplier.apply(directoryReader); - assert nonClosingReaderWrapper instanceof NonClosingReaderWrapper; - } + NonClosingReaderWrapper nonClosingReaderWrapper = new NonClosingReaderWrapper(engineSearcher.getDirectoryReader()); DirectoryReader reader = readerWrapper.apply(nonClosingReaderWrapper); if (reader != nonClosingReaderWrapper) { if (reader.getReaderCacheHelper() != openSearchDirectoryReader.getReaderCacheHelper()) { @@ -2588,7 +2541,6 @@ public void close(String reason, boolean flushEngine, boolean deleted) throws IO changeState(IndexShardState.CLOSED, reason); } } finally { - nonClosingReaderWrapperCache.clear(); final Indexer engine = this.currentEngineReference.getAndSet(null); try { if (engine != null && flushEngine) { @@ -4595,9 +4547,7 @@ public void beforeRefresh() {} public void afterRefresh(boolean didRefresh) { if (!didRefresh) return; // Use the engine directly (not IndexShard.acquireSearcher) so that we do NOT - // go through IndexShard.wrapSearcher / nonClosingReaderWrapperSupplier. - // Going through the shard-level wrapper would create entries in the - // nonClosingReaderWrapperCache that callers do not expect. + // go through IndexShard.wrapSearcher. try ( Engine.Searcher searcher = applyOnEngine( getIndexer(), @@ -6369,16 +6319,6 @@ public static T applyOnEngine(Indexer indexer, Function applier) } } - // Visible for testing - Function nonClosingReaderWrapperSupplier() { - return nonClosingReaderWrapperSupplier; - } - - // Visible for testing - ConcurrentHashMap nonClosingReaderWrapperCache() { - return nonClosingReaderWrapperCache; - } - // Visible for testing Object getEngineMutex() { return engineMutex; diff --git a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java index 6b2d88d448226..3b4cedcd180a0 100644 --- a/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java +++ b/server/src/test/java/org/opensearch/index/shard/IndexShardTests.java @@ -5514,54 +5514,6 @@ public void testPeriodicFlushTaskExecutesFlush() throws Exception { closeShards(primary); } - public void testCacheWrapperReader() throws IOException { - Settings settings = Settings.builder() - .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT) - .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0) - .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) - .put(IndexSettings.INDEX_PERIODIC_FLUSH_INTERVAL_SETTING.getKey(), "1s") - .build(); - - IndexMetadata metadata = IndexMetadata.builder("test") - .putMapping("{ \"properties\": { \"foo\": { \"type\": \"text\"}}}") - .settings(settings) - .primaryTerm(0, 1) - .build(); - - CheckedFunction wrapper = reader -> reader; - - IndexShard primary = newShard(new ShardId(metadata.getIndex(), 0), true, "n1", metadata, wrapper); - recoverShardFromStore(primary); - indexDoc(primary, "_doc", "0", "{\"foo\" : \"bar\"}"); - primary.flush(new FlushRequest()); - - try ( - Engine.SearcherSupplier searcherSupplier = primary.acquireSearcherSupplier(); - Engine.Searcher searcher = searcherSupplier.acquireSearcher("foo") - ) { - DirectoryReader directoryReader = searcher.getDirectoryReader(); - Engine.Searcher wrap = IndexShard.wrapSearcher(searcher, wrapper, primary.nonClosingReaderWrapperSupplier()); - wrap.close(); - assertEquals(1, primary.nonClosingReaderWrapperCache().size()); - DirectoryReader nonClosingReaderWrapper = primary.nonClosingReaderWrapperCache().get(directoryReader); - assertNotNull(nonClosingReaderWrapper); - - // use the cache - wrap = IndexShard.wrapSearcher(searcher, wrapper, primary.nonClosingReaderWrapperSupplier()); - wrap.close(); - assertEquals(1, primary.nonClosingReaderWrapperCache().size()); - DirectoryReader newNonClosingReaderWrapper = primary.nonClosingReaderWrapperCache().get(directoryReader); - assertEquals(nonClosingReaderWrapper, newNonClosingReaderWrapper); - - // not use the cache - wrap = IndexShard.wrapSearcher(searcher, wrapper, null); - assertNotEquals(wrap, newNonClosingReaderWrapper); - wrap.close(); - } - closeShards(primary); - assertTrue(primary.nonClosingReaderWrapperCache().isEmpty()); - } - /** * Verifies that {@code isRemoteSegmentStoreInSync} uses {@code getCatalogSnapshot()} (the unified * catalog API) rather than the legacy {@code getSegmentInfosSnapshot()}. After indexing and refreshing,