This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new cda19d472d9 [FLINK-40595][s3] Preserve multipart uploads during
recoverable stream disposal
cda19d472d9 is described below
commit cda19d472d9b3cabe3eed97107674f7fa5540f6d
Author: Mate Czagany <[email protected]>
AuthorDate: Mon Sep 28 13:55:18 2026 +0200
[FLINK-40595][s3] Preserve multipart uploads during recoverable stream
disposal
---
docs/content.zh/docs/deployment/filesystems/s3.md | 6 +++
docs/content/docs/deployment/filesystems/s3.md | 6 +++
flink-filesystems/flink-s3-fs-native/README.md | 6 +++
.../NativeS3RecoverableFsDataOutputStream.java | 40 +++++++-------
.../NativeS3RecoverableFsDataOutputStreamTest.java | 62 +++-------------------
5 files changed, 48 insertions(+), 72 deletions(-)
diff --git a/docs/content.zh/docs/deployment/filesystems/s3.md
b/docs/content.zh/docs/deployment/filesystems/s3.md
index 45f87ec9c51..27d4a77bde2 100644
--- a/docs/content.zh/docs/deployment/filesystems/s3.md
+++ b/docs/content.zh/docs/deployment/filesystems/s3.md
@@ -153,6 +153,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 b [...]
+#### Cleaning Up Unfinished Uploads
+
+The Native S3 FileSystem never aborts a multipart upload on its own. An upload
that is in progress may be referenced by a checkpoint or savepoint, and Flink
cannot tell whether that is the case when a stream is closed, cancelled, or
fails to commit. Aborting such an upload would make the referencing checkpoint
or savepoint unrecoverable. Uploads that are no longer needed (for example
after a job is cancelled and never restored, or after a failed commit)
therefore remain in the bucket and [...]
+
+**You must configure an S3 lifecycle rule that aborts incomplete multipart
uploads** on every bucket written through this file system. Without such a
rule, abandoned uploads accumulate indefinitely. See the [S3 documentation on
aborting incomplete multipart
uploads](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html)
for how to configure the rule. Choose a retention period long enough for
uploads to finish and for jobs to recover, includi [...]
+
#### Setup
To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to
the `plugins` directory:
diff --git a/docs/content/docs/deployment/filesystems/s3.md
b/docs/content/docs/deployment/filesystems/s3.md
index 9f84f9e6349..40756c5d95c 100644
--- a/docs/content/docs/deployment/filesystems/s3.md
+++ b/docs/content/docs/deployment/filesystems/s3.md
@@ -153,6 +153,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 b [...]
+#### Cleaning Up Unfinished Uploads
+
+The Native S3 FileSystem never aborts a multipart upload on its own. An upload
that is in progress may be referenced by a checkpoint or savepoint, and Flink
cannot tell whether that is the case when a stream is closed, cancelled, or
fails to commit. Aborting such an upload would make the referencing checkpoint
or savepoint unrecoverable. Uploads that are no longer needed (for example
after a job is cancelled and never restored, or after a failed commit)
therefore remain in the bucket and [...]
+
+**You must configure an S3 lifecycle rule that aborts incomplete multipart
uploads** on every bucket written through this file system. Without such a
rule, abandoned uploads accumulate indefinitely. See the [S3 documentation on
aborting incomplete multipart
uploads](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html)
for how to configure the rule. Choose a retention period long enough for
uploads to finish and for jobs to recover, includi [...]
+
#### Setup
To use the Native S3 FileSystem, copy the JAR file from the `opt` directory to
the `plugins` directory:
diff --git a/flink-filesystems/flink-s3-fs-native/README.md
b/flink-filesystems/flink-s3-fs-native/README.md
index 7f45ffe13ab..4686186099e 100644
--- a/flink-filesystems/flink-s3-fs-native/README.md
+++ b/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
+
+The Native S3 FileSystem never aborts a multipart upload on its own. An upload
that is in progress may be referenced by a checkpoint or savepoint, and Flink
cannot tell whether that is the case when a stream is closed, cancelled, or
fails to commit. Aborting such an upload would make the referencing checkpoint
or savepoint unrecoverable. Uploads that are no longer needed (for example
after a job is cancelled and never restored, or after a failed commit)
therefore remain in the bucket and [...]
+
+**You must configure an S3 lifecycle rule that aborts incomplete multipart
uploads** on every bucket written through this file system. Without such a
rule, abandoned uploads accumulate indefinitely. See the [S3 documentation on
aborting incomplete multipart
uploads](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html)
for how to configure the rule. Choose a retention period long enough for
uploads to finish and for jobs to recover, includi [...]
+
## Supported URI Schemes
This module supports both `s3://` and `s3a://` URI schemes:
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
index 4afe81c3a9c..79084c23eab 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java
@@ -51,6 +51,12 @@ import java.util.concurrent.locks.ReentrantLock;
* {@link ReentrantLock} guards the critical sections in {@link #close()},
{@link
* #closeForCommit()}, and {@link #persist()} to ensure safe cleanup of local
resources without
* corrupting S3 state.
+ *
+ * <p>This stream never aborts its multipart upload. The upload may be
referenced by a {@link
+ * NativeS3Recoverable} in a retained checkpoint or savepoint, and the stream
cannot know whether
+ * that is the case. Aborting it would make the referencing snapshot
unrecoverable. Abandoned
+ * uploads are expected to be removed by an S3 lifecycle rule for incomplete
multipart uploads, as
+ * with the Hadoop-based S3 file system.
*/
@NotThreadSafe
class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStream {
@@ -199,7 +205,7 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
// 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. nextPartNumber is only advanced on success.
NativeS3ObjectOperations.UploadPartResult result =
s3AccessHelper.uploadPart(
key, uploadId, nextPartNumber, currentTempFile,
currentPartSize);
@@ -233,11 +239,20 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
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.
+ // Only local resources are released. The upload is
deliberately left open: a
+ // previous persist() may have handed it out in a recoverable
that a completed
+ // checkpoint references, and aborting it would break recovery
from that
+ // checkpoint. See the class-level Javadoc.
closed = true;
+ LOG.warn(
+ "Failed to close multipart upload for commit (key={},
uploadId={}). The "
+ + "upload is left open because a checkpoint
may still reference "
+ + "it; an S3 lifecycle rule is expected to
remove it if unused.",
+ key,
+ uploadId,
+ e);
try {
- tryAbortUploadAndReleaseResources();
+ releaseLocalResources();
} catch (IOException cleanup) {
e.addSuppressed(cleanup);
}
@@ -285,15 +300,15 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
try {
if (!closed) {
closed = true;
- tryAbortUploadAndReleaseResources();
+ releaseLocalResources();
}
} finally {
unlock();
}
}
- /** Aborts the multipart upload and releases local resources on the best
effort basis. */
- private void tryAbortUploadAndReleaseResources() throws IOException {
+ /** Closes the local buffer and deletes only the local temp file. */
+ private void releaseLocalResources() throws IOException {
IOException collected = null;
if (currentOutputStream != null) {
try {
@@ -309,17 +324,6 @@ class NativeS3RecoverableFsDataOutputStream extends
RecoverableFsDataOutputStrea
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 (collected != null) {
throw collected;
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java
index 99a6824b9af..c2d63ac9c32 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java
@@ -61,7 +61,7 @@ class NativeS3RecoverableFsDataOutputStreamTest {
}
@Test
- void closeForCommitAbortsMultipartUploadWhenPartUploadFails() throws
Exception {
+ void closeForCommitKeepsMultipartUploadWhenPartUploadFails() throws
Exception {
s3.failUploadPart = true;
assertThatThrownBy(stream::closeForCommit)
@@ -69,56 +69,19 @@ class NativeS3RecoverableFsDataOutputStreamTest {
.hasMessageContaining("injected uploadPart failure");
assertThat(s3.abortAttempts)
- .as("closeForCommit must abort the upload on failure")
- .isEqualTo(1);
- assertThat(s3.openMultipartUploads)
- .as("the multipart upload must not leak after a failed commit")
- .doesNotContainKey(uploadId);
- assertThat(countLocalFilesIn(tmp)).as("the local temp file must be
cleaned up").isZero();
- }
-
- @Test
- void closeForCommitSurfacesAbortFailureWhenBothUploadAndAbortFail() throws
Exception {
- s3.failUploadPart = true;
- s3.failAbortMultiPartUpload = true;
-
- assertThatThrownBy(stream::closeForCommit)
- .isInstanceOf(IOException.class)
- .hasMessageContaining("injected uploadPart failure")
- .satisfies(
- t ->
- assertThat(t.getSuppressed())
- .as("the abort failure must be
surfaced, not swallowed")
- .anySatisfy(
- s ->
- assertThat(s)
-
.hasMessageContaining(
-
"injected abort failure")));
-
- assertThat(s3.abortAttempts).isEqualTo(1);
- }
-
- @Test
- void closeSurfacesAbortFailureInsteadOfSwallowingIt() throws Exception {
- s3.failAbortMultiPartUpload = true;
-
- assertThatThrownBy(stream::close)
- .isInstanceOf(IOException.class)
- .hasMessageContaining("injected abort failure");
-
- assertThat(s3.abortAttempts).isEqualTo(1);
- assertThat(countLocalFilesIn(tmp))
- .as("local resources are still released even when the abort
fails")
+ .as("closeForCommit must not abort the upload, a snapshot may
reference it")
.isZero();
+ assertThat(s3.openMultipartUploads).containsKey(uploadId);
+ assertThat(countLocalFilesIn(tmp)).as("the local temp file must be
cleaned up").isZero();
}
- /** An abnormal {@code close()} aborts the upload and releases local
state. */
+ /** An abnormal {@code close()} releases local state but keeps the upload.
*/
@Test
- void closeAbortsMultipartUploadOnAbnormalClose() throws Exception {
+ void closeKeepsMultipartUploadOnAbnormalClose() throws Exception {
stream.close();
- assertThat(s3.abortAttempts).isEqualTo(1);
- assertThat(s3.openMultipartUploads).doesNotContainKey(uploadId);
+ assertThat(s3.abortAttempts).isZero();
+ assertThat(s3.openMultipartUploads).containsKey(uploadId);
assertThat(countLocalFilesIn(tmp)).isZero();
}
@@ -130,10 +93,6 @@ class NativeS3RecoverableFsDataOutputStreamTest {
assertThatThrownBy(failingStream::close)
.isInstanceOf(IOException.class)
.hasMessageContaining("injected temp-file delete failure");
-
- assertThat(s3.abortAttempts)
- .as("abort is still attempted despite delete failure")
- .isEqualTo(1);
}
@Test
@@ -261,7 +220,6 @@ class NativeS3RecoverableFsDataOutputStreamTest {
racingStream.close();
- assertThat(s3.abortAttempts).isEqualTo(1);
assertThat(countLocalFilesIn(dir)).isZero();
}
@@ -314,7 +272,6 @@ class NativeS3RecoverableFsDataOutputStreamTest {
final Map<String, Map<Integer, byte[]>> openMultipartUploads = new
HashMap<>();
boolean failUploadPart = false;
- boolean failAbortMultiPartUpload = false;
boolean deletePartFileAfterUpload = false;
int abortAttempts = 0;
int uploadPartAttempts = 0;
@@ -393,9 +350,6 @@ class NativeS3RecoverableFsDataOutputStreamTest {
@Override
public void abortMultiPartUpload(String key, String uploadId) throws
IOException {
abortAttempts++;
- if (failAbortMultiPartUpload) {
- throw new IOException("injected abort failure for uploadId: "
+ uploadId);
- }
openMultipartUploads.remove(uploadId);
}
}
