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

docs/content.zh/docs/deployment/filesystems/s3.md

@@ -157,6 +157,12 @@ The legacy configuration key `s3.path.style.access` is still supported as a fall
 
 The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2 completely removing the dependency on Hadoop. It is registered under the schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref "docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`), server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role assumption, entropy injection, and bulk copy via S3TransferManager.
 
+#### Cleaning Up Unfinished Uploads
+
+Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored.
+
+To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance]({{< ref "docs/connectors/datastream/filesystem" >}}#s3-specific).
+
 #### Setup
 
 To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to the `plugins` directory:

docs/content/docs/deployment/filesystems/s3.md

@@ -157,6 +157,12 @@ The legacy configuration key `s3.path.style.access` is still supported as a fall
 
 The Native S3 FileSystem is a pure-Java implementation built on the AWS SDK v2 completely removing the dependency on Hadoop. It is registered under the schemes *s3://* and *s3a://*. It provides a drop-in replacement for the Presto and Hadoop implementations, supporting checkpointing, the [FileSink]({{< ref "docs/connectors/datastream/filesystem" >}}) (via `RecoverableWriter`), server-side encryption (SSE-S3, SSE-KMS), cross-account access via IAM role assumption, entropy injection, and bulk copy via S3TransferManager.
 
+#### Cleaning Up Unfinished Uploads
+
+Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored.
+
+To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance]({{< ref "docs/connectors/datastream/filesystem" >}}#s3-specific).
+
 #### Setup
 
 To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to the `plugins` directory:

flink-filesystems/flink-s3-fs-native/README.md

@@ -6,6 +6,12 @@ This module provides a native S3 filesystem implementation for Apache Flink usin
 
 The Native S3 FileSystem is a direct implementation of Flink's FileSystem interface using AWS SDK v2, without Hadoop dependencies. It provides exactly-once semantics for checkpointing and file sinks through S3 multipart uploads.
 
+### Cleaning Up Unfinished Uploads
+
+Flink keeps unfinished S3 uploads that may be needed to restore a job from a checkpoint or savepoint. These uploads can remain after a job stops if it is never restored.
+
+To clean up unused uploads, configure an S3 lifecycle rule for incomplete multipart uploads. Choose a retention period long enough for uploads to finish and for jobs to recover, including any planned downtime. S3 measures this period from when an upload starts. Cleaning up uploads too soon can prevent recovery from older checkpoints or savepoints. This rule does not remove other temporary files. See the [S3-specific FileSink guidance](../../docs/content/docs/connectors/datastream/filesystem.md#s3-specific).
+
 ## Supported URI Schemes
 
 This module supports both `s3://` and `s3a://` URI schemes:

flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java

@@ -78,14 +78,26 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea
 
     private volatile boolean closed;
 
+    // Recovered uploads and uploads handed out by persist() may belong to retained snapshots.
+    private boolean uploadMayBeReferenced;
+
     public NativeS3RecoverableFsDataOutputStream(
             NativeS3ObjectOperations s3AccessHelper,
             String key,
             String uploadId,
             String localTmpDir,
             long minPartSize)
             throws IOException {
-        this(s3AccessHelper, key, uploadId, localTmpDir, minPartSize, new ArrayList<>(), 0L, null);
+        this(
+                s3AccessHelper,
+                key,
+                uploadId,
+                localTmpDir,
+                minPartSize,
+                new ArrayList<>(),
+                0L,
+                null,
+                false);
     }
 
     public NativeS3RecoverableFsDataOutputStream(
@@ -98,6 +110,29 @@ public NativeS3RecoverableFsDataOutputStream(
             long numBytesInParts,
             File incompleteTailFile)
             throws IOException {
+        this(
+                s3AccessHelper,
+                key,
+                uploadId,
+                localTmpDir,
+                minPartSize,
+                existingParts,
+                numBytesInParts,
+                incompleteTailFile,
+                true);
+    }
+
+    private NativeS3RecoverableFsDataOutputStream(
+            NativeS3ObjectOperations s3AccessHelper,
+            String key,
+            String uploadId,
+            String localTmpDir,
+            long minPartSize,
+            List<PartETag> existingParts,
+            long numBytesInParts,
+            File incompleteTailFile,
+            boolean uploadMayBeReferenced)
+            throws IOException {
         this.s3AccessHelper = s3AccessHelper;
         this.key = key;
         this.uploadId = uploadId;
@@ -108,6 +143,7 @@ public NativeS3RecoverableFsDataOutputStream(
         this.nextPartNumber = existingParts.size() + 1;
         this.currentPartSize = 0;
         this.closed = false;
+        this.uploadMayBeReferenced = uploadMayBeReferenced;
 
         if (incompleteTailFile != null) {
             resumeFromIncompleteTail(incompleteTailFile);
@@ -199,7 +235,8 @@ private void uploadCurrentPart() throws IOException {
 
         // Do not delete the temp file if uploadPart fails: propagate the original exception
         // unmasked and let the cleanup path (close() or the closeForCommit() failure handler)
-        // delete it and abort the upload. nextPartNumber is only advanced on success.
+        // delete it and abort the upload if it is not needed for recovery.
+        // nextPartNumber is only advanced on success.
         NativeS3ObjectOperations.UploadPartResult result =
                 s3AccessHelper.uploadPart(
                         key, uploadId, nextPartNumber, currentTempFile, currentPartSize);
@@ -233,8 +270,8 @@ public Committer closeForCommit() throws IOException {
                         new NativeS3Recoverable(
                                 key, uploadId, new ArrayList<>(completedParts), numBytesInParts);
             } catch (IOException e) {
-                // The commit failed after the multipart upload had been created and parts may
-                // already have been uploaded. Abort it so it does not leak as an orphan upload.
+                // The failed commit may leave uploaded parts behind. Abort the upload to avoid an
+                // orphan only if it is not needed for recovery.
                 closed = true;
                 try {
                     tryAbortUploadAndReleaseResources();
@@ -267,6 +304,7 @@ public RecoverableWriter.ResumeRecoverable persist() throws IOException {
                 incompletePartLength = currentPartSize;
             }
 
+            uploadMayBeReferenced = true;
             return new NativeS3Recoverable(
                     key,
                     uploadId,
@@ -292,7 +330,9 @@ public void close() throws IOException {
         }
     }
 
-    /** Aborts the multipart upload and releases local resources on the best effort basis. */
+    /**
+     * Releases local resources and aborts uploads that cannot be referenced by recoverable state.
+     */
     private void tryAbortUploadAndReleaseResources() throws IOException {
         IOException collected = null;
         if (currentOutputStream != null) {
@@ -309,16 +349,18 @@ private void tryAbortUploadAndReleaseResources() throws IOException {
                 collected = ExceptionUtils.firstOrSuppressed(e, collected);
             }
         }
-        try {
-            s3AccessHelper.abortMultiPartUpload(key, uploadId);
-        } catch (IOException e) {
-            LOG.warn(
-                    "Failed to abort multipart upload (key={}, uploadId={}); it may be left as an "
-                            + "orphan upload in S3. Propagating the failure to the caller.",
-                    key,
-                    uploadId,
-                    e);
-            collected = ExceptionUtils.firstOrSuppressed(e, collected);
+        if (!uploadMayBeReferenced) {
+            try {
+                s3AccessHelper.abortMultiPartUpload(key, uploadId);
+            } catch (IOException e) {
+                LOG.warn(
+                        "Failed to abort multipart upload (key={}, uploadId={}); it may be left as an "
+                                + "orphan upload in S3. Propagating the failure to the caller.",
+                        key,
+                        uploadId,
+                        e);
+                collected = ExceptionUtils.firstOrSuppressed(e, collected);
+            }
         }
         if (collected != null) {
             throw collected;