[FLINK-39786][s3]closeForCommit() leaks orphan multipart uploads on upload failure

[Samrat002]

What is the purpose of the change

If uploadCurrentPart() throws after closed=true, the method exits without aborting the MPU and close() short-circuits on closed. The upload sits in S3 indefinitely. Even when close() does run, an abortMultiPartUpload failure is only WARN with a comment hoping bucket lifecycle rules clean up. Practically, most buckets don't have AbortIncompleteMultipartUpload configured.

Brief change log

Fix the leak

Verifying this change

1. Unit test
2. Run the flink job using s3 for fs.

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

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

Documentation

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

---

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

• Yes (claude sonnet for UT)

Generated-by: [Tool Name and Version]

[flinkbot]

CI report:

• 8538838 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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