[core] Close the data writer even when the changelog writer fails

[PDGGK]

Purpose

MergeTreeWriter#flushWriteBuffer closes two rolling writers in one finally, in sequence and unguarded:

final RollingFileWriter<KeyValue, DataFileMeta> changelogWriter =
        changelogProducer == ChangelogProducer.INPUT
                ? writerFactory.createRollingChangelogFileWriter(0)
                : null;
final RollingFileWriter<KeyValue, DataFileMeta> dataWriter =
        writerFactory.createRollingMergeTreeFileWriter(0, FileSource.APPEND);

try {
    writeBuffer.forEach(...);
} finally {
    writeBuffer.clear();
    if (changelogWriter != null) {
        changelogWriter.close();
    }
    dataWriter.close();          // <- skipped when the line above throws
}

RollingFileWriterImpl#close aborts its own files and rethrows (RollingFileWriterImpl:181-197), so changelogWriter.close() genuinely can throw, and dataWriter.close() is then never reached.

dataWriter is a local. It appears at its declaration, in the forEach, in that close(), and in the result() loop that runs after the finally — it is stored in no field and registered nowhere:

• createRollingMergeTreeFileWriter (KeyValueFileWriterFactory:136-155) does nothing but return new RollingFileWriterImpl<>(...).
• the factory's only cleanup hook, abortManagedBlobWrites() (:125-129), delegates to blobExternalizer and nothing else.
• MergeTreeWriter#close (:345-384) walks newFiles, newFilesChangelog, compactAfter and compactChangelog — all of which are populated after the finally, so on this path none of them can contain the abandoned writer's output.
• the outer handler, AbstractFileStoreWrite#close, reaches MergeTreeWriter#close, which holds no reference to the local either.

So once the method unwinds, nothing in the process can close or abort that writer: its open stream stays open, and any file it had already rolled is left in the bucket directory unreferenced by any snapshot.

Reachable whenever changelog-producer = input — that is the only configuration in which changelogWriter is non-null.

Separately, a finally that throws discards the exception in flight, so a genuine failure inside writeBuffer.forEach is replaced by the close failure rather than carrying it.

What changes

Both writers close through the idiom this repo already uses for exactly this, four call sites away in AbstractFileStoreWrite:381:

IOUtils.closeAll(changelogWriter, dataWriter);

IOUtils.closeAll skips nulls (IOUtils:199), calls close() on every element, and rethrows the first failure with the rest attached as suppressed. It throws Exception, which flushWriteBuffer already declares.

Blast radius

On the success path both writers are still closed exactly once and in the same order, and changelogWriter.result() / dataWriter.result() are reached identically. Only the failure path differs. MergeTreeTestBase (both sort engines, 22 tests) passes unchanged.

Test

MergeTreeWriterCloseFailureTest drives a TraceableFileIO whose streams close and then report failure, which is what a full disk or a rejected object-store finalize looks like. It writes two records through a changelog-producer = input writer, expects prepareCommit to fail, and then asserts that no output stream is left open — TraceableFileIO.openOutputStreams(...), the same check SingleFileWriterTest:189 uses.

Reverting to the original finally fails it, and names the leak:

Expecting empty but was:
  [OutStream{file=.../bucket-0/data-e878b92f-30b8-42bb-a050-21aedb514d35-1.parquet, stack=...

That is the data writer's stream, still open after the method returned.

MergeTreeWriterCloseFailureTest plus MergeTreeTestBase$MergeTreeTestWithLoserTree and $MergeTreeTestWithMinHeap: 23 tests, 0 failures. spotless:apply and checkstyle:check on paimon-core are clean.

Note

This is the same defect class as #9227, and the comment that PR left at AbstractFileStoreWrite:377-380 describes it — "closing them in a plain loop meant the first failure abandoned every writer behind it". This is another instance of it, in a method the earlier change did not reach.

API and Format

No change to any public signature, option or on-disk format.

[JingsongLi]

+1