[FLINK-40595][s3] Preserve multipart uploads during recoverable stream disposal

[mateczagany]

What is the purpose of the change

Prevent the native S3 writer from aborting multipart uploads that may be needed for recovery. Previously, disposing an in-progress file during task shutdown or cancellation could invalidate checkpoint state, causing recovery to fail with NoSuchUpload.

Brief change log

• Preserve uploads after successful persistence and when resuming from recoverable state
• Document cleanup of abandoned uploads and its implications for recovery

Verifying this change

This change added tests and can be verified as follows:

• Added unit tests covering failed first and subsequent persistence, disposal of an empty recovered stream, and concurrent persistence and disposal
• Extended existing S3 integration tests to dispose the original stream before recovery
• Added integration tests covering disposal of recovered streams, recovery after failed commit preparation, and cleanup of unpersisted uploads

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes — recovery of files written through the native S3 connector
• The S3 file system connector: yes

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Codex GPT-6

[flinkbot]

CI report:

• b802f41 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[gaborgsomogyi]

This adds a new field, a new constructor overload, and a concurrency edge case (persist()/close() race on uploadMayBeReferenced) just to distinguish "never persisted" from "persisted" uploads. I'm not yet convinced that we need 400 lines to fix this. Why don't we just skip abort in close()?

[mateczagany]

@gaborgsomogyi I would not mind skipping abort all together in close(), that was my original idea, and that's how the current Hadoop implementation works. But I had no other idea to keep the fix of FLINK-39786 working.

Also, persist() and close() hold the same lock, so I don't think there is a race on the flag.

[gaborgsomogyi]

I've tested this against your FLINK-39786 concern. Removing the abort entirely from plain close() (keeping it only in closeForCommit()'s existing failure catch) does not touch 39786 fix. Its own tests (closeForCommitAbortsMultipartUploadWhenPartUploadFails, closeForCommitSurfacesAbortFailureWhenBothUploadAndAbortFail) exercise closeForCommit()'s catch block, which stays untouched.

What does break are 4 tests asserting plain close() current abort behavior:

closeAbortsMultipartUploadOnAbnormalClose
closeSurfacesAbortFailureInsteadOfSwallowingIt
closeSurfacesTempFileDeletionFailure
closeDeletesTempFileRemovedDuringCleanup

Which specific 39786 scenario do you have in mind or something in closeForCommit()?

[mateczagany]

Yes, removing abort from close() does help and leaves FLINK-39786’s closeForCommit() failure handling unchanged.

However, aborting in closeForCommit() when a completed checkpoint already references the upload, even in the catch block can cause further issues with FileSink:

• The sink uploads a full part, e.g. 5 MiB, under upload ID U1
• A checkpoint completes while the file remains open. snapshotState() calls persist() (FileWriterBucket.java:217), saving a recoverable referencing U1
• More records arrive, leaving a small tail buffered locally
• The rolling policy triggers closing the part file (FileWriterBucket.java:231), reaching closeForCommit()
• Uploading that final tail fails with an S3 service error after retries. For example, HTTP 503. The error becomes an IOException, and the failure handler aborts U1 (NativeS3RecoverableFsDataOutputStream.java:235). The abort succeeds
• The task restarts from the completed checkpoint. Recovery reuses U1; its next part upload fails because U1 was aborted. S3 reports NoSuchUpload

So the full fix would be either to:

• Keep the work of FLINK-39786 and add flag uploadMayBeReferenced that's checked in closeForCommit(). If it's true, don't abort MPU. We don't have to use the flag in close(), we can simply always skip abort there
• Simply never abort MPUs, but I feel like that would go against the spirit of FLINK-39786. This is the path I would have chosen before I saw that ticket

[gaborgsomogyi]

Makese sense.

Stepping back: I've now found two separate cases where selective abort corrupts recoverable state (close() after persist(), and closeForCommit()-failure after persist()). Any such scheme relies on proving "no recoverable references this uploadId" from process-local state, and a power outage alone proves that guarantee is never airtight. A lifecycle policy is mandatory regardless of what the code does.

Given that, proactive abort only buys marginally faster cleanup within a retention window that must already be conservative for correctness reasons. That's not worth the recurring correctness risk of a hand-rolled flag/state machine that's already needed patching twice.

Proposal: drop abort entirely, in both close() and closeForCommit()'s failure catch, matching the Hadoop connector. Rely on the documented S3 lifecycle rule for cleanup.

@Samrat002 WDYT since you've added that?