DirectoryReader.open and openIfChanged with an ExecutorService leak the segment readers of accepted tasks when a later submit is rejected

Since 10.4.0, DirectoryReader.open and openIfChanged accept an ExecutorService and open one segment per task (#15428). StandardDirectoryReader.createSegmentReaders submits all tasks in one loop with no try/catch, and closes readers on failure only in its second loop, the one that collects the results. When the executor accepts some tasks and then rejects a submit, as a full bounded ThreadPoolExecutor with the default AbortPolicy does, RejectedExecutionException leaves the method from the first loop: the readers the accepted tasks built are never collected and never closed, and the caller cannot close them because it never receives them. With two one-document segments and an executor that runs the first task and rejects the second, MockDirectoryWrapper then refuses to close: "there are still 2 open files: {_1.fdx=1, _1.fdt=1}". For a refresh, a reused segment reader is left at refCount 2 instead of 1. Interrupting the opening thread while it waits leaks that task's reader the same way. The open fails loudly and search results are unaffected; the leak is silent and repeats on every failed attempt. Every release from 10.4.0 to 10.6.0 and main is affected. A fix that skips tasks that have not started and waits for the running ones passes the test and the closest suites.

• Executor overloads: eight public DirectoryReader methods take an ExecutorService; six of them can reach StandardDirectoryReader.createSegmentReaders (StandardDirectoryReader.java:187-247): DirectoryReader.open(Directory, ExecutorService) and five siblings (open with a leaf sorter or an IndexCommit, openIfChanged with or without an IndexCommit; DirectoryReader.java:71, :106-107, :164, :217-221, :269-270, :295-296 at 645d4fe14a20), added in 10.4.0 by #15428. The initial open reaches it at :96, a refresh through the reopen factory at :181. The other two, the NRT overloads that also take an IndexWriter (DirectoryReader.java:383-384, :425-429), never use the executor: they return null or call writer.getReader (StandardDirectoryReader.java:444-475). On an NRT reader only openIfChanged with an IndexCommit uses the executor, by reopening to the commit (StandardDirectoryReader.java:420-430, :496-505).
• Rejection: ExecutorService.submit throws RejectedExecutionException when it cannot take a task, for example a ThreadPoolExecutor whose threads and queue are full, with the JDK's default AbortPolicy, or any executor after shutdown().

1. The executor branch (:195-228) first submits one task per segment, newest first, in a loop with no try/catch (:197-207, executor.submit at :203). A task opens a new SegmentReader (:292-296) or, for a refresh, takes a reference on the old reader of an unchanged segment ("this reader will be shared between the old and the new one, so we must incRef it", :319-323).
2. Then it collects the results with Future.get() into readers[] (:208-217). If a task failed it keeps going, "to allow us to close created readers" (:213-214), and the cleanup decRefWhileSuppressingException(firstException, readers) (:221) releases every collected reader.
3. A RejectedExecutionException from the submit at :203 leaves the method before step 2. The futures of the tasks already accepted are dropped without being awaited, their readers never reach readers[], and the cleanup never runs. The caller gets only the exception.
4. Second trigger: an InterruptedException from get() is caught together with ExecutionException (:212), and the loop moves on. The task it was waiting for still runs; if it builds a reader, nobody stores or closes it.
5. The branch without an executor (:229-245) creates readers inside try and releases them on any Throwable (:241-242).

• Who is hit. Applications that call one of the six executor overloads that can reach this code with an executor that can reject. No Lucene module passes an executor here itself: IndexWriter uses the NRT open, and the replicator calls the reopen factory with null.
• Default vs opt-in. Opt-in. Without an executor the sequential branch runs, which cleans up. Common pools such as Executors.newFixedThreadPool queue without bound and do not reject while running.
• What triggers it. (a) A bounded pool with AbortPolicy that is full when the open submits its tasks: a stock ThreadPoolExecutor(1, 1, 0, ms, new SynchronousQueue<>()) whose worker is busy reproduced it. (b) An executor shut down while an open or refresh is in flight, after it accepted the first task (if it rejects the first submit, nothing leaks). (c) Secondary: the opening thread is interrupted, or cancelled with Future.cancel(true), while it waits.
• How much. Per failed attempt, the reader of every task accepted before the rejection: its open files and segment core; for a refresh, an extra reference on each reused segment reader, so closing the old reader no longer closes it. Lucene never releases them. From the code, linear in failed attempts (a retry loop was not run) and in accepted tasks (one leaked reader in the minimal test, two in the first finder's three-segment open test); any index with at least two segments. Whether garbage collection later reclaims file handles was not measured.
• What does not happen. No wrong results, no index change, no hang. The open or refresh fails loudly, and after a failed refresh the old reader keeps working.

Two segments are the fewest that allow "accept one, reject the next"; running the accepted task in the calling thread makes the outcome independent of timing. The single oracle is that the directory closes with nothing open; the same test is the regression test. Assertions on and off give the same failure, and every seed tried fails (DEADBEEF, 1, C0FFEE); only the names of the leaked files change with the randomized codec, and they always belong to _1.

./gradlew :lucene:core:checkGoogleJavaFormat flags only the test's one comment line (113 characters), which is too wide for the formatter. The independent reproducer and the cross-model reviews also reproduced the leak with a stock single-worker ThreadPoolExecutor over a SynchronousQueue whose worker was held inside the first task; that variant needs a gate in the directory to be deterministic, so it stays supporting evidence. All runs were on Linux aarch64 with OpenJDK 25.0.3; the bug is plain control flow, so no platform dependence is expected.

TestExecutorOpenRejectedSubmitLeak > testRejectedSubmitReleasesReaderOfAcceptedTask FAILED
    java.lang.RuntimeException: MockDirectoryWrapper: cannot close: there are still 2 open files: {_1.fdx=1, _1.fdt=1}
        at org.apache.lucene.tests.store.MockDirectoryWrapper.close(MockDirectoryWrapper.java:884)
        at org.apache.lucene.index.TestExecutorOpenRejectedSubmitLeak.testRejectedSubmitReleasesReaderOfAcceptedTask(TestExecutorOpenRejectedSubmitLeak.java:75)

        Caused by:
        java.lang.RuntimeException: unclosed IndexInput: _1.fdx
            at org.apache.lucene.tests.store.MockDirectoryWrapper.openInput(MockDirectoryWrapper.java:843)
            ...
            at org.apache.lucene.index.SegmentReader.<init>(SegmentReader.java:95)
            at org.apache.lucene.index.StandardDirectoryReader.createOrReuseSegmentReader(StandardDirectoryReader.java:296)
            at org.apache.lucene.index.StandardDirectoryReader.lambda$createSegmentReaders$0(StandardDirectoryReader.java:205)
            at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:328)
            at org.apache.lucene.index.TestExecutorOpenRejectedSubmitLeak$1.execute(TestExecutorOpenRejectedSubmitLeak.java:46)
            at java.base/java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:149)
            at org.apache.lucene.index.StandardDirectoryReader.createSegmentReaders(StandardDirectoryReader.java:203)
1 test completed, 1 failed

(gradle, main 645d4fe14a20, seed DEADBEEF, assertions on; assertions off gives the same message. The "Caused by"
trace shows the leaked file was opened inside the accepted task, called from the submit at line 203. On the
10.4.0-10.6.0 release jars the same test reports {_1.cfs=1}: the codec picked for the seed differs.)

• The code says so. "If there is an exception creating the reader we still process the rest of the completed futures to allow us to close created readers" (StandardDirectoryReader.java:213-214).
• The author said so. In the #15428 review a reviewer suggested TaskExecutor; the author answered: "I went with an ExecutorService here directly to ensure we could gracefully close out the readers in the event of an exception" (BryceKan3, 2025-11-18).
• The same method without an executor releases every reader on any Throwable (:229-245).
• Upstream tests. TestDirectoryReader.testExecutorServiceExceptionHandling (:1300-1316) passes a shut-down executor, expects a RuntimeException, then closes a newDirectory(), which fails on open files. It rejects the first submit, so no task was accepted. testExecutorServicePartialFailure (:1318-1335) fails a task through a failed future, not a rejected submit. Neither covers this path.
• Ownership. IndexReader.incRef (IndexReader.java:164-174): "Be sure to always call a corresponding decRef, in a finally clause; otherwise the reader may never be closed." The caller cannot do that for readers it never received.
• Rejection is ordinary. Lucene's own TaskExecutor treats rejecting executors as legitimate: "execute directly on the current thread in case of rejection to ensure a rejecting executor only reduces parallelism and does not result in failure" (search/TaskExecutor.java:58-67, #13622).

The strongest case for "intended" (the devil's-advocate verifier and the intended-lens reviewer): the javadoc of the executor overloads says only "executorService used to open segment readers in parallel" and @throws IOException (DirectoryReader.java:64-70), and the executor is the caller's own object, so its rejections are the caller's problem.

Why it fails: the caller can release only what it receives, and the failed open returns nothing; the code comment, the author's review answer, the sequential branch and upstream's rejecting-executor test all require cleanup. For the interrupt trigger the case is stronger: FSDirectory (store/FSDirectory.java:77-82) tells applications that use Thread.interrupt() or Future#cancel to switch to RAFDirectory. That note is about channels closed by an interrupt during I/O, while this code catches InterruptedException on purpose and carries on. Verdict: rejection high, interrupt medium.

Both fixes were run through gradle in a scratch worktree of 645d4fe14a20 (OpenJDK 25.0.3; seeds DEADBEEF and 5EED1234) with the minimal repro, the closest suites (TestDirectoryReader, TestDirectoryReaderReopen, TestIndexReaderClose, TestReaderClosed) and the format check, and standalone with the supporting tests. They differ in what a failed open waits for.

In createSegmentReaders, wrap the submit loop in try/catch (RuntimeException). Each task first claims an AtomicBoolean; if the claim fails, it returns null without opening anything. In the collect loop, once the open has failed (a rejected submit, a failed task or an interrupt), the opener claims each task that has not started and skips it; it waits only for tasks that already started. Then it releases every collected reader as today, restores the interrupt flag and rethrows. +45/-17 lines in StandardDirectoryReader.java.

@@ -31,6 +31,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.IOContext;
@@ -194,31 +195,58 @@
     final SegmentReader[] readers = new SegmentReader[sis.size()];
     if (executor != null) {
       List<Future<SegmentReader>> futures = new ArrayList<>();
-      for (int i = sis.size() - 1; i >= 0; i--) {
-        SegmentCommitInfo commitInfo = sis.info(i);
-        SegmentReader oldReader =
-            getOldSegmentReader(
-                oldReaders, previousSegmentReaders.get(commitInfo.info.name), commitInfo);
-        futures.add(
-            executor.submit(
-                () ->
-                    createOrReuseSegmentReader(
-                        commitInfo, oldReader, sis.getIndexCreatedVersionMajor())));
-      }
+      // A task only creates a reader if it claims its flag first. Once the open has failed, we
+      // claim the flags of the tasks that have not started yet, so we never wait for them.
+      List<AtomicBoolean> claims = new ArrayList<>();
       RuntimeException firstException = null;
+      try {
+        for (int i = sis.size() - 1; i >= 0; i--) {
+          SegmentCommitInfo commitInfo = sis.info(i);
+          SegmentReader oldReader =
+              getOldSegmentReader(
+                  oldReaders, previousSegmentReaders.get(commitInfo.info.name), commitInfo);
+          AtomicBoolean claim = new AtomicBoolean();
+          futures.add(
+              executor.submit(
+                  () ->
+                      claim.compareAndSet(false, true)
+                          ? createOrReuseSegmentReader(
+                              commitInfo, oldReader, sis.getIndexCreatedVersionMajor())
+                          : null));
+          claims.add(claim);
+        }
+      } catch (RuntimeException e) {
+        // e.g. RejectedExecutionException: readers of tasks that already started are closed below
+        firstException = e;
+      }
+      boolean interrupted = false;
       for (int i = 0; i < futures.size(); i++) {
-        try {
-          readers[sis.size() - 1 - i] = futures.get(i).get();
-        } catch (ExecutionException | InterruptedException e) {
-          // If there is an exception creating the reader we still process
-          // the rest of the completed futures to allow us to close created readers
-          if (firstException == null) firstException = new RuntimeException(e);
+        while (true) {
+          if (firstException != null && claims.get(i).compareAndSet(false, true)) {
+            break; // the task has not started, and now it never creates a reader
+          }
+          try {
+            readers[sis.size() - 1 - i] = futures.get(i).get();
+            break;
+          } catch (InterruptedException e) {
+            // Skip the task if it has not started yet, else wait for the reader it creates
+            interrupted = true;
+            if (firstException == null) firstException = new RuntimeException(e);
+          } catch (ExecutionException e) {
+            // If there is an exception creating the reader we still process
+            // the rest of the completed futures to allow us to close created readers
+            if (firstException == null) firstException = new RuntimeException(e);
+            break;
+          }
         }
       }
       if (firstException != null) {
         // If its a new reader decRef should close... otherwise old readers we will remove the new
         // refs that were created
         decRefWhileSuppressingException(firstException, readers);
+        if (interrupted) {
+          Thread.currentThread().interrupt();
+        }
         // Unwrap IOExceptions
         Throwable cause = firstException.getCause();
         if (cause instanceof ExecutionException && (cause).getCause() instanceof IOException) {

Trade-offs

• Strict and live. When open throws, every reader built has been released. Once the open has failed, the opener never waits for a task that has not started: a full or shut-down pool cannot make a failed open hang, and an interrupted opener returns once the running task finishes. It still waits for tasks that have started, as today's success path does.
• Cost. One AtomicBoolean and one compareAndSet per segment on the success path. It still calls executor.submit, so executors that override submit (as upstream's createExecutorServiceThatFailsOneTask does) see no change. No API change.
• Behaviour change. After a failure, still-queued tasks become no-ops. The cross-model reviewer's stock-pool test assumes its accepted task always opens a file; under B the opener may claim that task first, so the test fails that precondition ("accepted worker never opened a segment file") in its 3 iterations and the suite reports the pool threads it left behind; its other 9 runs pass. Not a leak.
• Scope. Catching RuntimeException also covers the IllegalStateException that getOldSegmentReader (:263-285) can throw inside the submit loop; an Error from submit still skips the cleanup, as today.

What the checks validated

Gradle, scratch worktree of 645d4fe14a20, removed afterwards: the minimal test fails stock and passes with B, assertions on and off; TestDirectoryReader (34), TestDirectoryReaderReopen (30), TestIndexReaderClose (3) and TestReaderClosed (1): 68 of 68 pass stock (seed DEADBEEF; the fact check also ran them stock at 5EED1234, 68 of 68) and with B (seeds DEADBEEF and 5EED1234); checkGoogleJavaFormat flags only the test's comment line, not the fix, and passes with B once the test is removed. Standalone: the minimal test (-ea, -da), the independent reproducer's interrupt test (-ea, -da; it hangs with fix A) and the first finder's three tests (two openIfChanged, one open) pass.

Wrap the submit loop in try/catch (RuntimeException), then wait for every future already submitted, also across an interrupt; release what was collected, restore the interrupt flag and rethrow. The study's sketch, re-done so that it keeps the method's comments. +35/-17 lines.

@@ -194,31 +194,49 @@
     final SegmentReader[] readers = new SegmentReader[sis.size()];
     if (executor != null) {
       List<Future<SegmentReader>> futures = new ArrayList<>();
-      for (int i = sis.size() - 1; i >= 0; i--) {
-        SegmentCommitInfo commitInfo = sis.info(i);
-        SegmentReader oldReader =
-            getOldSegmentReader(
-                oldReaders, previousSegmentReaders.get(commitInfo.info.name), commitInfo);
-        futures.add(
-            executor.submit(
-                () ->
-                    createOrReuseSegmentReader(
-                        commitInfo, oldReader, sis.getIndexCreatedVersionMajor())));
-      }
       RuntimeException firstException = null;
+      try {
+        for (int i = sis.size() - 1; i >= 0; i--) {
+          SegmentCommitInfo commitInfo = sis.info(i);
+          SegmentReader oldReader =
+              getOldSegmentReader(
+                  oldReaders, previousSegmentReaders.get(commitInfo.info.name), commitInfo);
+          futures.add(
+              executor.submit(
+                  () ->
+                      createOrReuseSegmentReader(
+                          commitInfo, oldReader, sis.getIndexCreatedVersionMajor())));
+        }
+      } catch (RuntimeException e) {
+        // e.g. RejectedExecutionException: still wait for the tasks that were already submitted,
+        // so that the readers they create are closed below
+        firstException = e;
+      }
+      boolean interrupted = false;
       for (int i = 0; i < futures.size(); i++) {
-        try {
-          readers[sis.size() - 1 - i] = futures.get(i).get();
-        } catch (ExecutionException | InterruptedException e) {
-          // If there is an exception creating the reader we still process
-          // the rest of the completed futures to allow us to close created readers
-          if (firstException == null) firstException = new RuntimeException(e);
+        while (true) {
+          try {
+            readers[sis.size() - 1 - i] = futures.get(i).get();
+            break;
+          } catch (InterruptedException e) {
+            // Keep waiting: the task may still create a reader that must be closed
+            interrupted = true;
+            if (firstException == null) firstException = new RuntimeException(e);
+          } catch (ExecutionException e) {
+            // If there is an exception creating the reader we still process
+            // the rest of the completed futures to allow us to close created readers
+            if (firstException == null) firstException = new RuntimeException(e);
+            break;
+          }
         }
       }
       if (firstException != null) {
         // If its a new reader decRef should close... otherwise old readers we will remove the new
         // refs that were created
         decRefWhileSuppressingException(firstException, readers);
+        if (interrupted) {
+          Thread.currentThread().interrupt();
+        }
         // Unwrap IOExceptions
         Throwable cause = firstException.getCause();
         if (cause instanceof ExecutionException && (cause).getCause() instanceof IOException) {

Trade-offs

Simplest. But it waits for every accepted task, also tasks that have not started. The independent reproducer's interrupt test, whose queued task can run only after open returns, hangs (timeout after 120 s, the opener parked in FutureTask.get). If the executor drops an accepted task, as shutdownNow() does with queued tasks without completing their futures (probe), a failed open would wait forever where today it throws at once. Today's success path already waits for every accepted task the same way.

What the checks validated

The same gradle runs as B: the minimal test passes with assertions on and off; 68 of 68 at seeds DEADBEEF and 5EED1234; the format check flags only the test's comment line, not the fix. Standalone: the cross-model reviewer's 12 tests (-ea, -da, 3 iterations) and the first finder's three tests pass.

The study's reviewers suggested cancelling the futures that have not started. The Future API cannot tell "not started" from "running".

Trade-offs

A probe on JDK 25: cancel(false) on a running FutureTask returns true, the task keeps running and produces its value, and get() then throws CancellationException: the value is lost. The reader of a running task would leak in a new way. B gets "cancel only if not started" from its claim flag.

As TaskExecutor does (search/TaskExecutor.java:58-67): catch the rejection and run the task in the opener's thread, so the open succeeds on a full pool.

Trade-offs

Changes behaviour that upstream's testExecutorServiceExceptionHandling asserts (a shut-down executor makes open throw), and does not fix the interrupt trigger, so it would still need A or B.

Method: formal, TLA+ model checking with TLC (a hunt unit, 2026-10-26), on a model of IndexReader reference counting and the StandardDirectoryReader open paths built from the code with a file:line map.

• With the faults rejectAt (a later submit is rejected) and interrupt (the opener is interrupted in Future.get), the NoLeak invariant fails; the model with the fix passes. Re-run on 645d4fe14a20 by a hunt unit: rejectAt3 violated at 93 distinct states (fixed: 77, OK), interrupt2 violated at 215 (fixed: 132, OK).
• The fault edges were added while modelling the code, so a strict reading could call the discovery semi-formal.
• Found independently by two other hunt units: one by code reading during an exception-path scan (both triggers), one by code reading while modelling commit discovery (interrupt trigger).
• A later hunt unit re-ran the model and reduced the counterexample to the one-test repro.