This is an automated email from the ASF dual-hosted git repository.
bamaer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 8d3dd4b653 return correct message when upload fails , fixes #9157
(#9158)
8d3dd4b653 is described below
commit 8d3dd4b6537f2760605584eb641a3a7920236e66
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Sat Oct 31 08:46:41 2026 +0100
return correct message when upload fails , fixes #9157 (#9158)
* return correct message when upload fails , fixes #9157
* fix remarks
---
.../vfs/s3/s3common/S3CommonPipedOutputStream.java | 51 +++++++++++++++++--
.../s3/s3common/S3CommonPipedOutputStreamTest.java | 58 ++++++++++++++++++++--
2 files changed, 102 insertions(+), 7 deletions(-)
diff --git
a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
index 16004ca4c4..30ae82ba69 100644
---
a/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
+++
b/plugins/tech/aws/src/main/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStream.java
@@ -26,10 +26,13 @@ import java.io.PipedOutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import lombok.Getter;
import lombok.Setter;
import org.apache.hop.core.logging.LogChannel;
@@ -53,6 +56,8 @@ public class S3CommonPipedOutputStream extends
PipedOutputStream {
public static final int DEFAULT_PART_SIZE = 5 * 1024 * 1024;
+ private static final long UPLOAD_RESULT_WAIT_SECONDS = 5;
+
private final ThreadPoolExecutor executor =
(ThreadPoolExecutor)
Executors.newFixedThreadPool(
@@ -103,13 +108,52 @@ public class S3CommonPipedOutputStream extends
PipedOutputStream {
@Override
public void write(int b) throws IOException {
initializeWrite();
- super.write(b);
+ try {
+ super.write(b);
+ } catch (IOException e) {
+ throw uploadFailure(e);
+ }
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
initializeWrite();
- super.write(b, off, len);
+ try {
+ super.write(b, off, len);
+ } catch (IOException e) {
+ throw uploadFailure(e);
+ }
+ }
+
+ /**
+ * A failed upload closes the reading end of the pipe, so the next write
fails with "Pipe closed".
+ * Report the upload failure instead. The upload closes the pipe as it
finishes, so its result
+ * follows shortly.
+ */
+ private IOException uploadFailure(IOException writeFailure) {
+ if (result == null) {
+ return writeFailure;
+ }
+ try {
+ if (Boolean.FALSE.equals(result.get(UPLOAD_RESULT_WAIT_SECONDS,
TimeUnit.SECONDS))) {
+ return new IOException(
+ BaseMessages.getString(PKG, "ERROR.S3MultiPart.ExceptionCaught")
+ + " ("
+ + bucketId
+ + "/"
+ + key
+ + ")",
+ writeFailure);
+ }
+ } catch (ExecutionException e) {
+ return new IOException(
+ "S3 multipart upload failed for " + bucketId + "/" + key,
e.getCause());
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (CancellationException | TimeoutException e) {
+ // No upload result to report: the write failure is all there is
+ }
+ return writeFailure;
}
@Override
@@ -243,7 +287,8 @@ public class S3CommonPipedOutputStream extends
PipedOutputStream {
}
LogChannel.GENERAL.logError(BaseMessages.getString(PKG,
"ERROR.S3MultiPart.Aborted"));
}
- returnVal = false;
+ // Rethrow so write() and close() report the S3 error, not just "Pipe
closed"
+ throw e;
}
return returnVal;
}
diff --git
a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
index 6b7ecfee50..ed9df4f5c9 100644
---
a/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
+++
b/plugins/tech/aws/src/test/java/org/apache/hop/vfs/s3/s3common/S3CommonPipedOutputStreamTest.java
@@ -17,6 +17,7 @@
package org.apache.hop.vfs.s3.s3common;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -26,6 +27,8 @@ import static org.mockito.Mockito.when;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import org.apache.hop.core.logging.HopLogStore;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
@@ -39,6 +42,12 @@ import
software.amazon.awssdk.services.s3.model.CreateMultipartUploadRequest;
*/
class S3CommonPipedOutputStreamTest {
+ @BeforeAll
+ static void initLogStore() {
+ // As in a normal Hop run: logging the upload failure must not throw
+ HopLogStore.init();
+ }
+
@Test
void closeIsInterruptibleWhenUploadStalls() throws Exception {
CountDownLatch uploadEntered = new CountDownLatch(1);
@@ -111,9 +120,50 @@ class S3CommonPipedOutputStreamTest {
// The upload may already have failed and closed the pipe: close() must
still report it
}
- assertThrows(
- IOException.class,
- out::close,
- "a failed S3 upload must be reported from close(), not silently
swallowed");
+ // The upload can fail before or after the bytes are written: write() or
close() reports it
+ IOException failure =
+ assertThrows(
+ IOException.class,
+ () -> {
+ out.write(new byte[] {1, 2, 3});
+ out.close();
+ },
+ "a failed S3 upload must be reported, not silently swallowed");
+ assertTrue(failure.getMessage().contains("test-bucket/test-key"),
failure.getMessage());
+ assertNotNull(failure.getCause(), "the S3 error must be the cause");
+ assertTrue(
+ failure.getCause().getMessage().contains("simulated S3 failure"),
+ failure.getCause().getMessage());
+ }
+
+ @Test
+ void writeAfterAFailedUploadReportsTheUploadFailure() throws Exception {
+ S3Client s3 = mock(S3Client.class);
+ when(s3.createMultipartUpload(any(CreateMultipartUploadRequest.class)))
+ .thenThrow(new RuntimeException("simulated S3 failure"));
+
+ S3CommonFileSystem fileSystem = mock(S3CommonFileSystem.class);
+ when(fileSystem.getS3Client()).thenReturn(s3);
+
+ S3CommonPipedOutputStream out =
+ new S3CommonPipedOutputStream(fileSystem, "test-bucket", "test-key");
+
+ // Keep writing until the failed upload has closed the pipe
+ IOException failure = null;
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (failure == null && System.nanoTime() < deadline) {
+ try {
+ out.write(1);
+ Thread.sleep(10);
+ } catch (IOException e) {
+ failure = e;
+ }
+ }
+ assertNotNull(failure, "writing after a failed upload must fail");
+ assertTrue(failure.getMessage().contains("test-bucket/test-key"),
failure.getMessage());
+ assertNotNull(failure.getCause(), "the S3 error must be the cause");
+ assertTrue(
+ failure.getCause().getMessage().contains("simulated S3 failure"),
+ failure.getCause().getMessage());
}
}
