You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ga...@apache.org on 2026/08/26 10:45:35 UTC
(flink) branch master updated: [FLINK-39874][s3] Make temp-file cleanup idempotent in NativeS3RecoverableFsDataOutputStream
This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi 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 5a07b9648fc [FLINK-39874][s3] Make temp-file cleanup idempotent in NativeS3RecoverableFsDataOutputStream
5a07b9648fc is described below
commit 5a07b9648fc29302043ba3b5604422f96c6fbe01
Author: Li Guo <al...@gmail.com>
AuthorDate: Wed Aug 26 03:45:27 2026 -0700
[FLINK-39874][s3] Make temp-file cleanup idempotent in NativeS3RecoverableFsDataOutputStream
---
.../NativeS3RecoverableFsDataOutputStream.java | 8 +-
.../writer/InMemoryNativeS3Operations.java | 13 +++
.../NativeS3RecoverableFsDataOutputStreamTest.java | 106 +++++++++++++++++++++
3 files changed, 123 insertions(+), 4 deletions(-)
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 2eb898b9e76..c883354f213 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
@@ -208,7 +208,7 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea
completedParts.add(new PartETag(result.getPartNumber(), result.getETag()));
numBytesInParts += currentPartSize;
- Files.delete(currentTempFile.toPath());
+ Files.deleteIfExists(currentTempFile.toPath());
}
@Override
@@ -226,7 +226,7 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea
if (currentPartSize > 0) {
uploadCurrentPart();
} else {
- Files.delete(currentTempFile.toPath());
+ Files.deleteIfExists(currentTempFile.toPath());
}
recoverable =
@@ -302,7 +302,7 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea
collected = ExceptionUtils.firstOrSuppressed(e, collected);
}
}
- if (currentTempFile != null && currentTempFile.exists()) {
+ if (currentTempFile != null) {
try {
deleteTempFile(currentTempFile);
} catch (IOException e) {
@@ -327,7 +327,7 @@ class NativeS3RecoverableFsDataOutputStream extends RecoverableFsDataOutputStrea
@VisibleForTesting
protected void deleteTempFile(File file) throws IOException {
- Files.delete(file.toPath());
+ Files.deleteIfExists(file.toPath());
}
private void lock() throws IOException {
diff --git a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
index 400f193291a..28d8edfeeef 100644
--- a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
+++ b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java
@@ -70,9 +70,18 @@ public final class InMemoryNativeS3Operations extends NativeS3ObjectOperations {
/** When {@code true}, {@link #uploadPart} throws to simulate a part-upload failure. */
public boolean failUploadPart = false;
+ /**
+ * When {@code true}, {@link #uploadPart} deletes the local part file after reading it,
+ * simulating an external cleaner that reaps {@code io.tmp.dirs} while the part is in flight.
+ */
+ public boolean deletePartFileAfterUpload = false;
+
/** When {@code true}, {@link #abortMultiPartUpload} throws to simulate an abort failure. */
public boolean failAbortMultiPartUpload = false;
+ /** Number of times {@link #uploadPart} was invoked, including failed attempts. */
+ public int uploadPartAttempts = 0;
+
/** Number of times {@link #abortMultiPartUpload} was invoked, including failed attempts. */
public int abortAttempts = 0;
@@ -100,6 +109,7 @@ public final class InMemoryNativeS3Operations extends NativeS3ObjectOperations {
public UploadPartResult uploadPart(
String key, String uploadId, int partNumber, File file, long length)
throws IOException {
+ uploadPartAttempts++;
if (failUploadPart) {
throw new IOException("injected uploadPart failure for uploadId: " + uploadId);
}
@@ -113,6 +123,9 @@ public final class InMemoryNativeS3Operations extends NativeS3ObjectOperations {
"part length mismatch: expected " + length + ", got " + data.length);
}
parts.put(partNumber, data);
+ if (deletePartFileAfterUpload) {
+ Files.delete(file.toPath());
+ }
return new UploadPartResult(partNumber, "etag-" + uploadId + "-" + partNumber);
}
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 676893aec03..35163cfe6f2 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
@@ -158,12 +158,118 @@ class NativeS3RecoverableFsDataOutputStreamTest {
assertThat(s3.committedObjects.get(KEY)).containsExactly(bytes('A', 5));
}
+ @Test
+ void closeForCommitDeletesTempFileOnSuccess() throws Exception {
+ assertThat(countLocalFilesIn(tmp))
+ .as("the pending part is buffered in a temp file")
+ .isOne();
+
+ stream.closeForCommit();
+
+ assertThat(s3.openMultipartUploads.get(uploadId))
+ .as("the commit must upload the pending part")
+ .containsOnlyKeys(1);
+ assertThat(countLocalFilesIn(tmp)).as("a successful commit deletes the temp file").isZero();
+ }
+
+ @Test
+ void closeForCommitDeletesAlreadyRemovedTempFile() throws Exception {
+ Path dir = tmp.resolve("empty-commit");
+ String uid = s3.startMultiPartUpload(KEY);
+ // No write(), so there is no pending part and closeForCommit() only deletes the temp file.
+ NativeS3RecoverableFsDataOutputStream emptyStream =
+ new NativeS3RecoverableFsDataOutputStream(
+ s3, KEY, uid, dir.toString(), MIN_PART_SIZE);
+ assertThat(countLocalFilesIn(dir)).as("the stream creates its temp file on open").isOne();
+
+ Files.delete(onlyFileIn(dir).toPath());
+
+ assertThat(emptyStream.closeForCommit()).as("the commit must still succeed").isNotNull();
+
+ assertThat(s3.abortAttempts).as("a healthy commit must not abort the upload").isZero();
+ assertThat(countLocalFilesIn(dir)).isZero();
+ }
+
+ @Test
+ void partUploadFailureLeavesTempFileForClose() throws Exception {
+ s3.failUploadPart = true;
+ assertThat(countLocalFilesIn(tmp))
+ .as("the pending part is buffered in a temp file")
+ .isOne();
+
+ // setUp() wrote 5 bytes; 5 more reach MIN_PART_SIZE and flush the part from write().
+ assertThatThrownBy(() -> stream.write(bytes('B', 5), 0, 5))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("injected uploadPart failure");
+
+ assertThat(countLocalFilesIn(tmp))
+ .as("a failed part upload keeps the temp file so the exception is not masked")
+ .isOne();
+
+ stream.close();
+
+ assertThat(countLocalFilesIn(tmp)).as("close() reclaims the temp file").isZero();
+ }
+
+ @Test
+ void partUploadDeletesAlreadyRemovedTempFile() throws Exception {
+ s3.deletePartFileAfterUpload = true;
+ File flushedFile = onlyFileIn(tmp);
+
+ stream.write(bytes('B', 5), 0, 5);
+
+ assertThat(s3.uploadPartAttempts).as("exactly one part was uploaded").isOne();
+ assertThat(flushedFile).as("the uploaded part file is gone").doesNotExist();
+ assertThat(onlyFileIn(tmp))
+ .as("write() rotated to a fresh temp file")
+ .isNotEqualTo(flushedFile);
+
+ stream.closeForCommit().commit();
+
+ assertThat(s3.committedObjects.get(KEY))
+ .as("the uploaded part is still committed")
+ .hasSize((int) MIN_PART_SIZE);
+ assertThat(countLocalFilesIn(tmp)).as("the commit deletes the rotated temp file").isZero();
+ }
+
+ @Test
+ void closeDeletesTempFileRemovedDuringCleanup() throws Exception {
+ Path dir = tmp.resolve("close-race");
+ String uid = s3.startMultiPartUpload(KEY);
+ // close() may run concurrently with the writer thread during cancellation, so the temp
+ // file may already be gone when close() deletes it.
+ NativeS3RecoverableFsDataOutputStream racingStream =
+ new NativeS3RecoverableFsDataOutputStream(
+ s3, KEY, uid, dir.toString(), MIN_PART_SIZE) {
+ @Override
+ protected void deleteTempFile(File file) throws IOException {
+ Files.delete(file.toPath());
+ super.deleteTempFile(file);
+ }
+ };
+ racingStream.write(bytes('A', 5), 0, 5);
+ assertThat(countLocalFilesIn(dir))
+ .as("the pending part is buffered in a temp file")
+ .isOne();
+
+ racingStream.close();
+
+ assertThat(s3.abortAttempts).isEqualTo(1);
+ assertThat(countLocalFilesIn(dir)).isZero();
+ }
+
private NativeS3RecoverableFsDataOutputStream newStream(
InMemoryNativeS3Operations ops, String uid) throws IOException {
return new NativeS3RecoverableFsDataOutputStream(
ops, KEY, uid, tmp.toString(), MIN_PART_SIZE);
}
+ private static File onlyFileIn(Path dir) {
+ File[] files = dir.toFile().listFiles();
+ assertThat(files).hasSize(1);
+ return files[0];
+ }
+
private NativeS3RecoverableFsDataOutputStream newFailingDeleteStream() throws IOException {
String uid = s3.startMultiPartUpload(KEY);
return new NativeS3RecoverableFsDataOutputStream(