NestDream commented on code in PR #28360: URL: https://github.com/apache/flink/pull/28360#discussion_r3707418883
########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java: ########## @@ -0,0 +1,197 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.fs.s3native.writer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the local temp-file handling of {@link NativeS3RecoverableFsDataOutputStream} on the + * part-upload failure and commit paths. + * + * <p>Cleanup after a failed {@code uploadPart()} is deliberately left to {@code close()} rather + * than done eagerly in {@code uploadCurrentPart()}, so that a delete failure cannot mask the + * original upload {@link IOException}. These tests pin that contract: the temp file survives the + * failed upload and is reclaimed by {@code close()}, and no {@code s3-part-*} file is left behind + * afterwards. + */ +class NativeS3RecoverableFsDataOutputStreamTest { + + private static final long MIN_PART_SIZE = 5L * 1024 * 1024; // 5 MB + private static final String KEY = "test/object"; + private static final String UPLOAD_ID = "test-upload-id"; + + /** + * When {@code uploadPart()} fails for a part flushed mid-stream (from {@code write()}), the + * temp file is intentionally retained so the upload exception propagates unmasked, and {@code + * close()} reclaims it. + */ + @Test + void uploadPartFailureFromWriteIsReclaimedByClose(@TempDir Path tmpDir) throws IOException { + FailingUploadHelper helper = new FailingUploadHelper(); + NativeS3RecoverableFsDataOutputStream stream = + new NativeS3RecoverableFsDataOutputStream( + helper, KEY, UPLOAD_ID, tmpDir.toString(), MIN_PART_SIZE); + + // Write >= minPartSize so write() flushes a part via uploadCurrentPart(), which fails. + byte[] payload = new byte[(int) MIN_PART_SIZE]; + + assertThatThrownBy(() -> stream.write(payload, 0, payload.length)) + .isInstanceOf(IOException.class); + + // Retained by design: uploadCurrentPart() must not delete on failure. + assertThat(findTempFile(tmpDir)) + .as("temp file is reclaimed by close(), not eagerly") + .isNotNull(); + + stream.close(); + + assertNoTempFilesRemain( + tmpDir, "after close() following an uploadPart() failure in write()"); + } + + /** + * Regression test for the commit path. {@code closeForCommit()} only sets {@code closed = true} + * after the final part has been uploaded, so when {@code uploadPart()} throws there the stream + * is still open and {@code close()} reclaims the temp file. Were {@code closed} set before the + * upload, {@code close()} would short-circuit on its {@code if (!closed)} guard and orphan the + * {@code s3-part-*} file in the shared {@code io.tmp.dirs} for the lifetime of the TaskManager. + */ + @Test + void closeForCommitUploadFailureIsReclaimedByClose(@TempDir Path tmpDir) throws IOException { + FailingUploadHelper helper = new FailingUploadHelper(); + NativeS3RecoverableFsDataOutputStream stream = + new NativeS3RecoverableFsDataOutputStream( + helper, KEY, UPLOAD_ID, tmpDir.toString(), MIN_PART_SIZE); + + // Write a small amount (< minPartSize) so no part is flushed during write(); the single + // pending part is uploaded only at commit time, where uploadPart() then fails. + stream.write(new byte[1024], 0, 1024); + + assertThatThrownBy(stream::closeForCommit).isInstanceOf(IOException.class); + + stream.close(); + + assertNoTempFilesRemain( + tmpDir, "after close() following an uploadPart() failure in closeForCommit()"); + } + + /** The temp file for a successfully uploaded part is deleted on the normal commit path. */ + @Test + void closeForCommitSuccessDeletesTempFile(@TempDir Path tmpDir) throws IOException { + NoopObjectOperations helper = new NoopObjectOperations(); + NativeS3RecoverableFsDataOutputStream stream = + new NativeS3RecoverableFsDataOutputStream( + helper, KEY, UPLOAD_ID, tmpDir.toString(), MIN_PART_SIZE); + + stream.write(new byte[1024], 0, 1024); + + assertThat(stream.closeForCommit()).isNotNull(); + assertNoTempFilesRemain(tmpDir, "after a successful commit upload"); + } + + /** + * {@code closeForCommit()} with no pending bytes takes the {@code else} branch and deletes the Review Comment: Trimmed in 3bd3968; the javadocs now describe the current contract only. ########## flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java: ########## @@ -280,9 +280,13 @@ public void close() throws IOException { cleanupException = ExceptionUtils.firstOrSuppressed(e, cleanupException); } } - if (currentTempFile != null && currentTempFile.exists()) { + if (currentTempFile != null) { try { - Files.delete(currentTempFile.toPath()); + // deleteIfExists() rather than exists() + delete(): write() and + // uploadCurrentPart() run without the lock and replace currentTempFile, + // so a concurrent cancellation close() could otherwise fail with + // NoSuchFileException between the exists() check and the delete(). + Files.deleteIfExists(currentTempFile.toPath()); Review Comment: Removed in 3bd3968. ########## flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java: ########## @@ -0,0 +1,197 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.fs.s3native.writer; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the local temp-file handling of {@link NativeS3RecoverableFsDataOutputStream} on the + * part-upload failure and commit paths. + * + * <p>Cleanup after a failed {@code uploadPart()} is deliberately left to {@code close()} rather + * than done eagerly in {@code uploadCurrentPart()}, so that a delete failure cannot mask the + * original upload {@link IOException}. These tests pin that contract: the temp file survives the + * failed upload and is reclaimed by {@code close()}, and no {@code s3-part-*} file is left behind + * afterwards. + */ +class NativeS3RecoverableFsDataOutputStreamTest { + + private static final long MIN_PART_SIZE = 5L * 1024 * 1024; // 5 MB + private static final String KEY = "test/object"; + private static final String UPLOAD_ID = "test-upload-id"; + + /** + * When {@code uploadPart()} fails for a part flushed mid-stream (from {@code write()}), the + * temp file is intentionally retained so the upload exception propagates unmasked, and {@code + * close()} reclaims it. + */ + @Test + void uploadPartFailureFromWriteIsReclaimedByClose(@TempDir Path tmpDir) throws IOException { Review Comment: Merged in 3bd3968 into one parameterized test, `uploadPartFailureIsReclaimedByClose`, with an `@EnumSource` over the two call sites. -- 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]
