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 <[email protected]>
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(
