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

[mateczagany]

What is the purpose of the change

Update the native S3 writer so that it never aborts multipart uploads as they may be needed for future recovery.

Since the stream cannot prove from process-local state that no retained checkpoint or savepoint references the upload, it now never aborts it, matching the Hadoop-based S3 file system. Abandoned uploads are left to an S3 lifecycle rule for incomplete multipart uploads.

Brief change log

• Remove the multipart upload abort from close() and from the closeForCommit() failure path
• Document cleanup of abandoned uploads and its implications for recovery

Verifying this change

This change is covered by existing and updated tests

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:

• d656565 Azure: SUCCESS

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?

[Samrat002]

Short history, The close() abort predates FLINK-39786. It's been there since the module's initial implementation. FLINK-39786 added the one in closeForCommit()'s catch block, for the case where uploadCurrentPart() throws while closed is already true, so the following close() no-ops and nothing aborts. It was a miss from my end.

The reason it was added because In FLIP-555, the idea was to use S3 lifecycle management to clean up orphaned uploads: "a lifecycle rule will get it eventually". In production, what we observed most buckets don't have AbortIncompleteMultipartUpload configured at all. This was causing substantial bill at higher scale.

I'm on board with dropping both. Few notes:

• uploadMayBeReferenced is a plain field while closed is volatile, and the constructor write marking recovered streams isn't under the lock, even though the javadoc allows close() from another thread. That's a third hole in the same invariant.

• The downside is asymmetric. Our commitAfterRecovery() is a bare commit(), unlike Hadoop's, which falls back to checking that the object exists at the expected length. A wrong abort then fails every restore of that checkpoint, and execution.checkpointing.num-retained defaults to 1.

We do lose FLINK-39786's case. A commit failure with no recoverable anywhere will now leak until the lifecycle rule fires. Worst case, if the bucket has no configured lifecycle rules. This is acceptable, as native-s3-fs does not claim to clean up orphan MPU. We can call this out in the documentation.

I am alligned with proposal:

1. The docs should say the lifecycle rule is required, not recommended.
2. If the complexity for cleanup is simple and doesnt not add too much overhead on flink part, then Mate's follow-up ticket for proper orphan cleanup is sensible. else we callout s3 bucket must have lifecycle enabled for lower storage cost.

[gaborgsomogyi]

@mateczagany did you have the chance to take a look at this? I think at least we 3 are on the same page that we handle the abandoned MPUs the same way as the Hadoop based connector which would reduce the complexity significantly.

[mateczagany]

Thank you @Samrat002 and @gaborgsomogyi . I have force-pushed to the branch to update the S3 recoverable stream to never abort multipart uploads.