Skip to content

[core] Close every writer and shut both executors down when one close fails - #9227

Open
PDGGK wants to merge 1 commit into
apache:masterfrom
PDGGK:fix-filestorewrite-close
Open

[core] Close every writer and shut both executors down when one close fails#9227
PDGGK wants to merge 1 commit into
apache:masterfrom
PDGGK:fix-filestorewrite-close

Conversation

@PDGGK

@PDGGK PDGGK commented Aug 14, 2026

Copy link
Copy Markdown
Contributor

Purpose

AbstractFileStoreWrite#close walks the writers map with a plain loop, so the first failure abandons everything behind it:

for (Map<Integer, WriterContainer<T>> bucketWriters : writers.values()) {
    for (WriterContainer<T> writerContainer : bucketWriters.values()) {
        writerContainer.writer.close();          // <- the only call here that can throw
        if (writerContainer.primaryKeyIndexMaintainer != null) {
            writerContainer.primaryKeyIndexMaintainer.close();
        }
    }
}
writers.clear();
if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) {
    lazyCompactExecutor.shutdownNow();
}
if (lazyPrimaryKeyIndexExecutor != null) {
    lazyPrimaryKeyIndexExecutor.shutdownNow();
}
if (compactionMetrics != null) {
    compactionMetrics.close();
}

RecordWriter#close is the only one of these that can throw — BucketedPrimaryKeyIndexMaintainer#close and CompactionMetrics#close are both declared void with no checked exception. So a single writer throwing takes out, in order:

  • every remaining writer in the map — and there is one per bucket per partition,
  • writers.clear(),
  • both shutdownNow() calls, leaving two thread pools alive for the rest of the process,
  • compactionMetrics.close().

The thread pools are the part that outlives the operation.

Tests

Four unit tests in AbstractFileStoreWriteCloseTest. Restoring the plain loop fails exactly three:

plain loop (current master) this PR
testCloseReleasesEveryWriterWhenAnEarlierOneThrows FAILED ok
testLaterFailuresRideAlongInsteadOfBeingDropped FAILED ok
testWriterMapIsClearedEvenWhenAWriterThrows FAILED ok
testCloseSucceedsWhenNoWriterThrows ok ok

The fourth covers the path where nothing throws and passes either way, so the first three are not trivially red.

mvn -pl paimon-core test -Dtest=AbstractFileStoreWriteCloseTest → 4 run, 0 failures. spotless:check and checkstyle:check on paimon-core are clean.

API and Format

No change to any public signature, option, or on-disk format. The only behaviour change is on the failure path: close() still throws the same first exception, but now the later failures are attached to it as suppressed rather than never happening, and the teardown after the loop always runs.

The writers go through IOUtils.closeAll, which is the helper Paimon already has for this — it closes all of them and rethrows the first failure with the rest suppressed. The tail moves into a finally so it runs whatever the writers did; since none of those four calls throws, the writer's failure is never replaced by one of them.

Same shape as SortMergeReaderWithMinHeap#close (#9163) and the lookup-store chain (#9172).

… fails

AbstractFileStoreWrite#close walked the writers map with a plain loop:

    for (Map<Integer, WriterContainer<T>> bucketWriters : writers.values()) {
        for (WriterContainer<T> writerContainer : bucketWriters.values()) {
            writerContainer.writer.close();
            ...
        }
    }
    writers.clear();
    if (lazyCompactExecutor != null && closeCompactExecutorWhenLeaving) {
        lazyCompactExecutor.shutdownNow();
    }
    if (lazyPrimaryKeyIndexExecutor != null) {
        lazyPrimaryKeyIndexExecutor.shutdownNow();
    }
    if (compactionMetrics != null) {
        compactionMetrics.close();
    }

RecordWriter#close is the only call in there that can throw --
BucketedPrimaryKeyIndexMaintainer#close and CompactionMetrics#close are
both declared void with no checked exception. So one writer throwing
takes out everything after it: every remaining writer in the map, the
writers.clear(), both executor shutdowns, and the metrics close. There
is one writer per bucket per partition, so a single bad writer can strand
a large number of them, and the two thread pools stay alive for the rest
of the process.

The writers now go through IOUtils.closeAll, which closes all of them and
rethrows the first failure with the others attached as suppressed. The
tail moves into a finally so it runs whatever the writers did; none of
those four calls throws, so the writer failure is never replaced by one
of them.

Same shape as SortMergeReaderWithMinHeap#close (apache#9163) and the
lookup-store chain (apache#9172), and it uses the helper paimon already has for
it rather than adding another.

Four unit tests. Restoring the plain loop fails three of them -- the
later writers are not closed, the suppressed failure is lost, and the
map is left populated. The fourth covers the path where nothing throws
and passes either way, so the first three are not trivially red.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant