OfflineSorter with an ExecutorService can leave temp files behind when sort() fails

OfflineSorter.sort() wraps its Directory in a TrackingDirectoryWrapper "So we can remove any partially written temp files on exception". With a non-null ExecutorService (the 8-argument constructor, since 6.6.0), each partition sort and each merge runs as a job on that executor, and a job creates its temp file only when it runs. When sort() fails, its catch blocks delete a snapshot of the files created so far and rethrow, without cancelling or waiting for the jobs already submitted, so jobs that run afterwards can leave files nobody deletes. Without an executor every job runs inside submit(), and the same failure leaves nothing. In the repro a truncated 1.1 MB input leaves foo_sort_1.tmp (1,016 sorted records) behind; with plain 1- and 2-thread pools, 32 to 39 of 320 injected I/O failures leaked files. Successful sorts are unaffected, the failed sort() returns at once and no result is wrong. The class is @lucene.internal and no in-tree caller on main passes an executor, so the reach is small. Waiting for the submitted jobs (after making queued ones no-ops) before the cleanup fixes it.

• Partition: as many input records as fit in the RAM buffer; reading one takes a partitionsInRAM permit (OfflineSorter.java:430-433).
• Job: a SortPartitionTask (sorts one partition into a temp file) or a MergePartitionsTask (merges up to maxTempFiles temp files), submitted with exec.submit(...).
• TrackingDirectoryWrapper: a Directory wrapper that records the name of every file created through it; getCreatedFiles() returns a copy of those names.

1. sort() starts with // So we can remove any partially written temp files on exception: and wraps the directory in a TrackingDirectoryWrapper (:295-296).
2. With exec != null the constructor keeps the caller's executor (:240-245); otherwise it uses a SameThreadExecutorService, whose execute just runs the job (:246-248; SameThreadExecutorService.java:34-37).
3. Each partition becomes a job, Callable<Partition> job = new SortPartitionTask(trackingDir, part); (:309), submitted with segments.add(exec.submit(job)); (:311); merges are submitted the same way (:394-398).
4. A job creates its temp file only when it runs: dir.createTempOutput(tempFileNamePrefix, "sort", ...) in SortPartitionTask.call (:630-631) and in MergePartitionsTask.call (:698-700).
5. On failure both catch blocks (:363-369) call IOUtils.deleteFilesSuppressingExceptions(..., trackingDir, trackingDir.getCreatedFiles()) and rethrow. That set is a copy taken at that moment (TrackingDirectoryWrapper.java:70-73), and nothing touches the futures in segments: no cancel, no get.
6. A job that had not yet created its file does so after the cleanup. Its name lands in the tracking set, but nobody reads the set again, and sort() gives the caller neither the generated names nor a way to wait for the jobs. A partition file stays unless a later merge consumes it (:749-753); a merge whose inputs the cleanup already deleted creates its output (:698-700), fails to open an input and leaves that output empty.

Without an executor every job has run before the failure is caught, so the snapshot is complete. A successful sort is fine too: it waits for the final job (:350, future.get() at :662), which waits for all its inputs.

Stock main, the repro's input and executor (a small probe, not the repro): partition 1 holds 1,016 records; right after sort() throws EOFException the directory holds only input_tmp_0.tmp and one job is still queued; once it runs, foo_sort_1.tmp is there for good: 1,018,048 bytes, 1,016 records, a valid footer. With exec == null the same input leaves only input_tmp_0.tmp.

Natural timing, no test gating: a harness that injects one IOException per run into the Directory or one of its inputs or outputs (at each of the first 40 such operations of a sort and at 40 randomly chosen later ones, 80 runs per configuration) leaked files in 32 (assertions on) and 39 (off) of 320 failed sorts with 1- and 2-thread FIFO pools, and in 0 of 160 without an executor. Other executed triggers: an injected read error 1,000 entries into partition 2, and an interrupt while sort() waits for a partition permit.

• Who is hit. Callers of the 8-argument constructor with a non-null ExecutorService, the documented way to sort concurrently. The default constructors are not affected.
• Reach. OfflineSorter is tagged @lucene.experimental and @lucene.internal. On main, hunspell SortingStrategy and the suggesters (SortedInputIterator, AnalyzingSuggester, FSTCompletionLookup) use the 2- or 3-argument constructors. In release 9.8.0 BPIndexReorderer (lucene/misc) passed its ForkJoinPool (when one was set) to this constructor; 9.9.0 dropped that use (#12712). A code search of the default branches of Elasticsearch, OpenSearch and Solr found no use of OfflineSorter.
• What happens. Per failed sort, at most one leftover <prefix>_sort_N.tmp for each job that had not yet created its file: at most maxPartitionsInRAM partition jobs plus any merge jobs already submitted. There can be fewer, because a merge that runs later deletes the partition files it consumes: with maxTempFiles 2, two queued partition jobs and their queued merge left one file, the merge's output. A leftover partition file is a complete sorted copy of the records it held (1,016 in the repro); a merge whose inputs the cleanup had already deleted fails and leaves an empty output file with no footer (both cases executed by the fact check's probes). The orphaned jobs also keep their CPU, I/O and partition buffer until they finish. The documented maxPartitionsInRAM * ramBufferSize bound is per sorter and still holds: a later sort() on the same sorter waits for those permits, and can wait forever if the executor never runs the orphaned jobs (for example after shutdownNow() dropped them; code reading).
• What does not happen. No wrong or partial result (sort() throws), the failed sort() does not hang, no index file change; a successful sort is unaffected. Assertions on or off make no difference.
• Likelihood. Rare: a sort must fail (disk error, damaged input, interrupt) while a job is in flight; under natural timing about 1 in 10 injected failures leaked.

The input is 1,100 records of 1,000 bytes, then a record header that claims 32,767 bytes and has none, then a valid footer. BufferSize.megabytes(1) makes partition 1 take 1,016 records. DeferredExecutor runs a queued job only when the test calls runQueued() or when someone calls get() on its future; the second rule lets a fix that waits for submitted jobs complete instead of hanging. ByteBuffersDirectory numbers temp files with one counter, so the input is input_tmp_0.tmp and the leaked file foo_sort_1.tmp. The test sits in org.apache.lucene.util to match Lucene's test layout and uses no package-private member. As written it has no ASF license header and one comment line is too long for google-java-format; a copy with both fixed passes :lucene:core:check.

As the regression test: the same test, for example as a method in TestOfflineSorter.

cp TestOfflineSorterFailureCleanup.java $LUCENE/lucene/core/src/test/org/apache/lucene/util/
cd $LUCENE && ./gradlew :lucene:core:test --tests org.apache.lucene.util.TestOfflineSorterFailureCleanup -Ptests.seed=DEADBEEF
# assertions off: add -Ptests.asserts=false (same result)

# gradle, main 645d4fe14a20, seed DEADBEEF (assertions on; the same with -Ptests.asserts=false)
TestOfflineSorterFailureCleanup > testFailedSortLeavesNoTempFiles FAILED
    java.lang.AssertionError: expected:<[input_tmp_0.tmp]> but was:<[foo_sort_1.tmp, input_tmp_0.tmp]>
        at __randomizedtesting.SeedInfo.seed([DEADBEEF:D7882769E7D7FE80]:0)
        at org.junit.Assert.fail(Assert.java:89)
        at org.junit.Assert.failNotEquals(Assert.java:835)
        at org.junit.Assert.assertEquals(Assert.java:120)
        at org.junit.Assert.assertEquals(Assert.java:146)
        at org.apache.lucene.util.TestOfflineSorterFailureCleanup.testFailedSortLeavesNoTempFiles(TestOfflineSorterFailureCleanup.java:49)

1 test completed, 1 failed

# release jars 10.6.0 (assertions on and off), 10.5.2 and 9.12.3 (on): the same assertion message

# step by step (a small probe with the repro's input and executor, stock main, assertions off)
PROBE executor sort() threw EOFException; records read into partition 1 = 1016; files right after sort() = [input_tmp_0.tmp]; tasks still queued = 1
PROBE executor files after the queue ran = [foo_sort_1.tmp, input_tmp_0.tmp]
PROBE executor leaked foo_sort_1.tmp: 1018048 bytes, 1016 records, footer ok
PROBE exec=null sort() threw EOFException; records read into partition 1 = 1016; files right after sort() = [input_tmp_0.tmp]; tasks still queued = 0
PROBE exec=null files after the queue ran = [input_tmp_0.tmp]

• Intent stated in the code. No javadoc promise: sort() says only "Sort input to a new temp file, returning its name" (:286). The code states the goal at :295, again at :353-354 ("We should be explicitly removing all intermediate files ourselves unless there is an exception"), and in the catch blocks (:363-369).
• The executor is presented as optional. Constructor javadoc: "If you pass a non-null ExecutorService then it will be used to run sorting operations that can be run concurrently" (:226-227); CHANGES: "OfflineSorter can now run concurrently if you pass it an optional ExecutorService" (lucene/CHANGES.txt:8312-8313). Nothing tells the caller to drain the executor or to delete files after a failure, and the caller never gets the generated names.
• History. LUCENE-4209 (CHANGES :15305-15308) fixed temp files left behind in the system temp directory. In LUCENE-6829 (#7887, 2015) Mike McCandless wrote: "I cutover to TrackingDirectory in OfflineSorter to manage deleting temp files on exception". When BKDWriter used this class, TestBKD.testWithExceptions checked "BKDWriter should fully clean up after itself" (7.7.3, TestBKD.java:413). LUCENE-7792 (db92a9efc2ee) moved file creation into jobs and kept the one-shot cleanup; its same-day follow-up 25f1dd2f9b2 "add try/finally to make sure semaphore is released on exceptions" shows the failure paths were meant to be handled. The #8843 thread never discusses a failure with jobs in flight.
• Tests. testBitFlippedOnInput1/2 use the 2-argument constructor and testBitFlippedOnPartition1/2 pass null (TestOfflineSorter.java:356, :414, :460, :528); randomExecutorServiceOrNull() is used only where the sort succeeds (:96, :120, :145, :551).
• Legal trigger. Public constructor, a standard executor, BufferSize.megabytes(1); the class detects corrupt input on purpose (compare testBitFlippedOnInput2) and catches the interrupt explicitly (:363).

The strongest case for "works as designed" (the intended-behaviour verifier and the cross-model reviews): the class is @lucene.experimental and @lucene.internal; the executor belongs to the caller, who can shut it down and delete files by prefix; the cleanup is best effort (deleteFilesSuppressingExceptions); there is no javadoc promise; and TaskExecutor waiting for sibling tasks after a failure (#12523) was labelled an enhancement. A maintainer may call the fix an improvement, or deprecate the executor option, which nobody in-tree has used since 9.9.0.

Why it fails: @lucene.internal is about API stability, not about cleanup being optional. Nothing documents a duty to drain the executor, and sort() gives the caller no completion signal and no list of the generated names: short of shutting down its own, possibly shared, executor or polling the Directory, it cannot tell when the orphaned jobs are done. The class already holds the futures and waits on them when the sort succeeds. "Best effort" covers a delete that fails, not a file created after the cleanup ran. The argument limits priority (low), not validity; this is an inference from the code, its comments and history, not a maintainer ruling.

Option A was run through gradle in a scratch worktree of 645d4fe14a20 and is the suggested option: validated against main's tests, not against every legal executor, so it needs a maintainer's design review. B, C and D are alternatives a maintainer may prefer; none was tried. No open PR touches OfflineSorter.java. The one-test repro passes under any option that leaves no file once the queued job has run.

sort() records every future it submits (partition and merge jobs) and a fresh aborted flag per call. Both catch blocks first call abortAndAwaitTasks(): set the flag, then get() every submitted future, ignoring the job's own failure, retrying on InterruptedException and restoring the interrupt flag at the end. Each job starts with: if aborted, return the permit (partition jobs) and throw CancellationException, so queued jobs create no file. Only then are trackingDir.getCreatedFiles() deleted.

diff --git a/lucene/core/src/java/org/apache/lucene/util/OfflineSorter.java b/lucene/core/src/java/org/apache/lucene/util/OfflineSorter.java
index a4d09ff674c..ffb9650a538 100644
--- a/lucene/core/src/java/org/apache/lucene/util/OfflineSorter.java
+++ b/lucene/core/src/java/org/apache/lucene/util/OfflineSorter.java
@@ -24,10 +24,12 @@ import java.util.Comparator;
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.lucene.codecs.CodecUtil;
 import org.apache.lucene.store.ChecksumIndexInput;
@@ -188,6 +190,10 @@ public class OfflineSorter {
   private final BufferSize ramBufferSize;
 
   SortInfo sortInfo;
+  // every task submitted by the current sort() call, so a failed sort can wait for them
+  private List<Future<Partition>> submitted;
+  // set when the current sort() call fails; tasks that have not started yet then do nothing
+  private AtomicBoolean aborted;
   private final int maxTempFiles;
   private final Comparator<BytesRef> comparator;
 
@@ -287,6 +293,8 @@ public class OfflineSorter {
   public String sort(String inputFileName) throws IOException {
 
     sortInfo = new SortInfo();
+    submitted = new ArrayList<>();
+    aborted = new AtomicBoolean();
     long startMS = System.currentTimeMillis();
 
     List<Future<Partition>> segments = new ArrayList<>();
@@ -308,7 +316,9 @@ public class OfflineSorter {
 
         Callable<Partition> job = new SortPartitionTask(trackingDir, part);
 
-        segments.add(exec.submit(job));
+        Future<Partition> future = exec.submit(job);
+        submitted.add(future);
+        segments.add(future);
         sortInfo.tempMergeFiles++;
         sortInfo.lineCount += part.count;
         levelCounts[0]++;
@@ -361,14 +371,40 @@ public class OfflineSorter {
       return result;
 
     } catch (InterruptedException ie) {
+      abortAndAwaitTasks();
       IOUtils.deleteFilesSuppressingExceptions(ie, trackingDir, trackingDir.getCreatedFiles());
       throw new ThreadInterruptedException(ie);
     } catch (Throwable t) {
+      abortAndAwaitTasks();
       IOUtils.deleteFilesSuppressingExceptions(t, trackingDir, trackingDir.getCreatedFiles());
       throw t;
     }
   }
 
+  /**
+   * Called when sort() fails: tasks that have not started yet become no-ops, and we wait for all
+   * submitted tasks to finish so that none of them creates a temp file after the cleanup.
+   */
+  private void abortAndAwaitTasks() {
+    aborted.set(true);
+    boolean interrupted = false;
+    for (Future<Partition> future : submitted) {
+      while (true) {
+        try {
+          future.get();
+          break;
+        } catch (InterruptedException _) {
+          interrupted = true;
+        } catch (ExecutionException | CancellationException _) {
+          break;
+        }
+      }
+    }
+    if (interrupted) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
   /**
    * Called on exception, to check whether the checksum is also corrupt in this source, and add that
    * information (checksum matched or didn't) as a suppressed exception.
@@ -395,7 +431,9 @@ public class OfflineSorter {
         new MergePartitionsTask(trackingDir, new ArrayList<>(segmentsToMerge));
 
     segmentsToMerge.clear();
-    segments.add(exec.submit(task));
+    Future<Partition> future = exec.submit(task);
+    submitted.add(future);
+    segments.add(future);
 
     sortInfo.tempMergeFiles++;
   }
@@ -627,6 +665,12 @@ public class OfflineSorter {
 
     @Override
     public Partition call() throws IOException {
+      if (aborted.get()) {
+        if (partitionsInRAM != null) {
+          partitionsInRAM.release();
+        }
+        throw new CancellationException("sort was aborted");
+      }
       try (IndexOutput tempFile =
               dir.createTempOutput(tempFileNamePrefix, "sort", IOContext.DEFAULT);
           ByteSequencesWriter out = getWriter(tempFile, part.buffer.size())) {
@@ -681,6 +725,9 @@ public class OfflineSorter {
 
     @Override
     public Partition call() throws IOException {
+      if (aborted.get()) {
+        throw new CancellationException("sort was aborted");
+      }
       long totalCount = 0;
       for (Future<Partition> segment : segmentsToMerge) {
         totalCount += getPartition(segment).count;

Trade-offs

• A failing sort() returns only after the running jobs have finished (bounded: at most maxPartitionsInRAM partition sorts plus running merges) and a worker has picked up each queued job (which returns at once). On a stalled or saturated caller-owned executor, or one shut down with shutdownNow() that drops queued jobs, the wait can be long or never end. The success path already waits on the executor the same way; today's failure path does not. For example, a sort() running on the only worker of its own executor already deadlocks on success (it waits for a job queued behind itself); with A it waits the same way on failure.
• The wait continues through further interrupts (the flag is restored), so a second interrupt cannot cut a failing sort short.
• Waiting before the snapshot also removes the unconfirmed getCreatedFiles() copy race.
• The new fields are set per call but live on the instance, and the jobs read the flag when they run, so A requires one sort() at a time per sorter (sortInfo is shared the same way; testThreadSafety uses one sorter per thread). Passing the call's list and flag to the jobs would drop that requirement. submitted keeps the finished futures until the next call and could be cleared at the end.
• The diff uses unnamed catch variables (catch (InterruptedException _), Java 22+): fine for main (minimum Java 25); a 10.x backport (minimum Java 21) needs named variables.
• No API or format change; the success path adds one list append per job.

What the checks validated

Gradle with JDK 25 in a scratch worktree of 645d4fe14a20 (removed afterwards):

• the repro fails on the stock build and passes with the fix, assertions on and off, also in its committable form (ASF header, long comment wrapped);
• TestOfflineSorter: seed DEADBEEF 15 tests, 0 failures (2 nightly skipped); seed 5EED1234 with -Ptests.iters=5 75 tests, 0 failures; with -Ptests.nightly=true 15 tests, 0 failures, 0 skipped (incl. the executor-using testLargerRandom and testFixedLengthHeap);
• :lucene:suggest:test: 31 classes, 236 tests, 0 failures, 7 skipped;
• :lucene:core:check -x test with the committable test: BUILD SUCCESSFUL (format, license headers, ecj lint, forbidden APIs); with the test exactly as in the repro it fails only on that file (no license header, one comment line too long);
• earlier, by the hunt: the fault-injection sweep with the fix leaked in 0 of 480 failed sorts (stock 32-39 of 320 with pools).

Not measured: performance.

Set the flag in the catch blocks as in A, and let every job check it again after writing its file and delete that file, also when the job fails because the cleanup removed its inputs.

Trade-offs

sort() throws at once, but the files disappear only when the jobs finish, sort() gives the caller no signal when that is, and the getCreatedFiles() copy race stays for a separate TrackingDirectoryWrapper fix (compare #13142). Not tried.

Future.cancel(false) on each queued job, then wait for the running ones before the cleanup.

Trade-offs

A cancelled SortPartitionTask never runs its finally, so its permit (taken in readPartition, released only by the job) would be lost and a later sort() on the same sorter could block, unless sort() releases it itself (the same permit accounting as LFV-384). Relies on the executor's futures honouring cancel. Not tried.

No in-tree caller has used the 8-argument constructor with an executor since 9.9.0, and the class is @lucene.internal.

Trade-offs

Simplest for main, but it leaves released versions as they are for the callers that use it, and it is an API decision for the maintainers.

Method: formal (TLA+ model checking, the tool's own counterexample). In the hunt's offline-sorter hunt (2026-10-26, main ea57360912ed), the hunting AI tool wrote a protocol model of sort(): the main thread, the partitionsInRAM semaphore, a FIFO executor with W workers, partition and merge jobs, the merge cascade, the final merge and the failure cleanup, with one injected failure (an I/O error, corruption or an interrupt) at any step. Record contents are not modelled.

• With no failure, TLC found no deadlock, every partition reaches the result exactly once, and only the result file remains.
• With one failure, the invariant FailureClean ("after a failed sort and once every job has finished, no temp file remains") is violated in all 12 configurations with 1 or 2 workers (shortest trace 9 states: partition 1 queued, the main thread fails, the cleanup deletes nothing, the queued job then creates its file) and holds in all 6 with 0 workers, the same-thread executor.
• The AI tool wrote the model and chose the invariant, the property the code's own comment states; TLC then found the failing schedule among every fault position and job schedule it explored, so the scenario was not hand-picked.

A fault-injection harness on the stock jar (one IOException per run, injected into the Directory or one of its inputs or outputs at each of the first 40 such operations and at 40 randomly chosen later ones; plain thread pools, no gating) confirmed it: 32 and 39 of 320 failed sorts leaked with 1- or 2-thread pools, 0 of 160 without an executor, 0 of 480 with the fix. The hunt's two-test JUnit followed; on 2026-10-29 it was replaced with the one deterministic test, because the interrupt test was not deterministic.