Izeren commented on code in PR #28951:
URL: https://github.com/apache/flink/pull/28951#discussion_r3767435807
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java:
##########
@@ -307,6 +290,38 @@ public void close() throws IOException {
}
}
+ /** Aborts the multipart upload and releases local resources on the best
effort basis. */
+ @Nullable
+ private IOException abortUploadAndReleaseResources(@Nullable IOException
primary) {
+ IOException collected = primary;
+ if (currentOutputStream != null) {
+ try {
+ currentOutputStream.close();
+ } catch (IOException e) {
+ collected = ExceptionUtils.firstOrSuppressed(e, collected);
Review Comment:
What happens when you add `firstOrSuppressed(e, null)`? Is it returning just
`e`?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/InMemoryNativeS3Operations.java:
##########
@@ -67,6 +67,15 @@ public final class InMemoryNativeS3Operations extends
NativeS3ObjectOperations {
/** uploadId → partNumber → uploaded bytes for in-flight MPUs. */
public final Map<String, Map<Integer, byte[]>> openMultipartUploads = new
HashMap<>();
+ /** When {@code true}, {@link #uploadPart} throws to simulate a
part-upload failure. */
+ public boolean failUploadPart = false;
Review Comment:
Do we have test for `fail to delete temp file`?
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStream.java:
##########
@@ -272,32 +280,7 @@ public void close() throws IOException {
try {
if (!closed) {
closed = true;
- IOException cleanupException = null;
- if (currentOutputStream != null) {
- try {
- currentOutputStream.close();
- } catch (IOException e) {
- cleanupException = ExceptionUtils.firstOrSuppressed(e,
cleanupException);
- }
- }
- if (currentTempFile != null && currentTempFile.exists()) {
- try {
- Files.delete(currentTempFile.toPath());
- } catch (IOException e) {
- cleanupException = ExceptionUtils.firstOrSuppressed(e,
cleanupException);
- }
- }
-
- try {
- s3AccessHelper.abortMultiPartUpload(key, uploadId);
- } catch (IOException e) {
- LOG.warn(
- "Multipart upload failed (key={}, uploadId={}). "
- + "S3 lifecycle rules should eventually
clean up the incomplete upload.",
- key,
- uploadId,
- e);
- }
+ IOException cleanupException =
abortUploadAndReleaseResources(null);
Review Comment:
This looks like a very unusual signature. Name of the method feels fully
detached from the fact that it acts that exception enricher alongside with
abort logic. Imho, it would be cleaner if this method would throw instead.
Maybe naming it as `tryAbortUploadAndReleaseResources` would be cleaner. Not
sure how to improve the fact that it takes exception as an argument.
Semantically it should take an argument `abortReason` which can be `standard`
vs `onFailure` where `onFailure` contains the exception. But I agree that the
last bit might look like an overkill for what you are trying to simplify and
javadocs explanation would suffice.
Note, that you don't need following:
```
if (cleanupException != null) {
throw cleanupException;
}
```
if throwing logic is inside the method
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/writer/NativeS3RecoverableFsDataOutputStreamTest.java:
##########
@@ -0,0 +1,188 @@
+/*
+ * 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.apache.flink.core.fs.RecoverableFsDataOutputStream;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test {@link NativeS3RecoverableFsDataOutputStream}. */
+class NativeS3RecoverableFsDataOutputStreamTest {
+
+ private static final String KEY = "out.txt";
+ private static final long MIN_PART_SIZE = 10L;
+
+ @TempDir Path tmp;
+
+ @Test
+ void closeForCommitAbortsMultipartUploadWhenPartUploadFails() throws
Exception {
+ InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
Review Comment:
Looks like a lot of boilerplate is repeated between tests? Can it be moved
to @BeforeEach?
```
InMemoryNativeS3Operations s3 = new InMemoryNativeS3Operations();
String uploadId = s3.startMultiPartUpload(KEY);
NativeS3RecoverableFsDataOutputStream stream = newStream(s3, uploadId);
stream.write(bytes('A', 5), 0, 5); // < MIN_PART_SIZE, so it is uploaded
during commit
```
--
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]
