[core] Take the stream position outside the finally so close always runs

[PDGGK]

Purpose

ObjectsFile#writeWithoutRolling reads the stream position inside the finally, ahead of the close:

PositionOutputStream out = fileIO.newOutputStream(path, false);
long pos;
try {
    try (FormatWriter writer = writerFactory.create(out, compression)) {
        while (records.hasNext()) {
            writer.addElement(serializer.toRow(records.next()));
        }
    }
} finally {
    pos = out.getPos();   // <- declared `throws IOException`
    out.close();          // <- skipped if it does
}

PositionOutputStream#getPos is public abstract long getPos() throws IOException. When it throws, out.close() is never reached and the stream leaks, and the finally's exception replaces whatever the try body was throwing — a finally that throws discards the exception in flight, and nothing is attached as suppressed.

The two failures are not independent. The try body usually throws because the underlying stream is already broken — a full disk, an object store rejecting the upload — and getPos on that same stream is then likely to throw as well. So the case where the position read fails is largely the same case where losing the write error hurts most.

What changes

The whole block becomes try-with-resources, so the stream is closed on every path and a close failure is attached to the original failure rather than promoted over it:

long pos;
try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
    try (FormatWriter writer = writerFactory.create(out, compression)) {
        while (records.hasNext()) {
            writer.addElement(serializer.toRow(records.next()));
        }
    }
    pos = out.getPos();
}
return Pair.of(path.getName(), pos);

The two blocks stay nested rather than folding into a single resource list: the position has to be read after the writer has flushed and before the stream itself closes, and one list would close the writer only on the way out, after getPos().

On a successful write the returned value is identical to before. On a failing write, the stream is closed and the write failure is the one that propagates, with any close failure among its suppressed exceptions.

An earlier revision of this PR moved pos onto the success path but left the finally { out.close(); } in place, and claimed that made the original exception propagate. It did not — @JingsongLi pointed out that a throwing out.close() would still replace it, which is the same defect the first paragraph above describes. The test below exists because of that.

Tests

ObjectsFileWriteFailureTest, two cases, driving a stub FileIO and FormatWriter:

• both fail — writer close throws writer close failed, stream close throws stream close failed. The propagated cause must be the writer failure, with the stream failure attached as suppressed.
• success path — the stream is still closed, and the returned position is the one getPos() reported.

The first case is what the previous revision got wrong; run against that revision it fails exactly as predicted:

Expecting message to be:
  "writer close failed"
but was:
  "stream close failed"

ObjectsFileWriteFailureTest and BinaryIndexManifestEntryTest: 4 tests, 0 failures. spotless:apply and checkstyle:check on paimon-core are clean.

API and Format

No change to any public signature, option, or on-disk format. Behaviour differs only on the failure path, where the stream is now closed and the write failure is the one that reaches the caller.

Note

#8927 also touches this method, in the catch (Throwable e) block below (deleteQuietly → deleteQuietlyIgnoringInterrupt). Different lines and a different concern, but worth flagging since they are close together.

[JingsongLi]

In ObjectsFile.writeWithoutRolling, if writer.close() (or the write operation itself) throws an exception and the subsequent out.close() in the finally block also throws an exception, the latter will still overwrite the original exception; thus, the claim in the PR that the "original exception propagates" is incorrect. It is recommended to use try-with-resources in the outer scope so that any exception thrown during closing is added to the list of suppressed exceptions.

[PDGGK]

You're right, and the claim in my description was wrong. A finally that throws discards the exception in flight, so out.close() failing would still have replaced the write failure — the very thing the PR said it fixed. Pushed a876331 using try-with-resources as you suggest.

long pos;
try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
    try (FormatWriter writer = writerFactory.create(out, compression)) {
        while (records.hasNext()) {
            writer.addElement(serializer.toRow(records.next()));
        }
    }
    pos = out.getPos();
}

I kept the two blocks nested rather than folding them into one resource list, because the position has to be read after the writer has flushed and before the stream closes; a single list would close the writer only on the way out, after getPos().

I also added a test, since the thing I got wrong was exactly the exception plumbing and I would rather it be pinned than argued. It drives a stub whose writer close and stream close both throw:

Expecting message to be:
  "writer close failed"
but was:
  "stream close failed"

That is the old try/finally under the new test — your case, reproduced. With a876331 the cause is writer close failed and stream close failed is attached to it as suppressed. A second case covers the success path, asserting the stream is still closed and the position is the one returned.

ObjectsFileWriteFailureTest and BinaryIndexManifestEntryTest: 4 tests, 0 failures. spotless:apply and checkstyle:check on paimon-core are clean.

Thanks for catching it — and for #9227.