Samrat002 commented on code in PR #29132:
URL: https://github.com/apache/flink/pull/29132#discussion_r4090215764
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java:
##########
@@ -233,11 +233,13 @@ 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.
+ // 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;
try {
- tryAbortUploadAndReleaseResources();
+ releaseLocalResources();
Review Comment:
IMO, adding a WARN log would help here. WDUT?
##########
docs/content.zh/docs/deployment/filesystems/s3.md:
##########
@@ -149,10 +149,20 @@ The legacy configuration key `s3.path.style.access` is
still supported as a fall
## Implementation Details
-### Native S3 FileSystem
+### Native S3 FileSystem (Experimental)
+
+{{< hint warning >}}
+**Experimental**: The Native S3 FileSystem is experimental in Flink 2.3. It is
functionally complete and has demonstrated strong performance in benchmarks.
+{{< /hint >}}
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
+
+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 are billed as storage until they are removed.
+
+**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. 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).
Review Comment:
[NIT] adding reference to
https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html
will be helpful
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java:
##########
@@ -285,15 +287,18 @@ public void close() throws IOException {
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 the local temp file. The multipart
upload is never
+ * aborted here; see the class-level Javadoc.
Review Comment:
```suggestion
* Closes the local buffer and deletes only the local temp file.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
