This is an automated email from the ASF dual-hosted git repository.
capistrant pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 1234058e3ab fix: Fix issue where RetryableS3OutputStream quietly
aborts an s3 multipart upload but doesn't throw to indicate failure (#20002)
1234058e3ab is described below
commit 1234058e3abbe547a2ae076e070e1f98d515d70c
Author: Lucas Capistrant <[email protected]>
AuthorDate: Fri Aug 14 05:21:17 2026 -0500
fix: Fix issue where RetryableS3OutputStream quietly aborts an s3 multipart
upload but doesn't throw to indicate failure (#20002)
---
.../storage/s3/output/RetryableS3OutputStream.java | 25 ++++++++++++++++
.../s3/output/RetryableS3OutputStreamTest.java | 35 +++++++++++++++-------
2 files changed, 49 insertions(+), 11 deletions(-)
diff --git
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/output/RetryableS3OutputStream.java
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/output/RetryableS3OutputStream.java
index 075f97ed1cc..b897fde4d12 100644
---
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/output/RetryableS3OutputStream.java
+++
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/output/RetryableS3OutputStream.java
@@ -22,6 +22,7 @@ package org.apache.druid.storage.s3.output;
import com.google.common.base.Stopwatch;
import com.google.common.io.CountingOutputStream;
import it.unimi.dsi.fastutil.io.FastBufferedOutputStream;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.RetryUtils;
import org.apache.druid.java.util.common.StringUtils;
@@ -295,9 +296,16 @@ public class RetryableS3OutputStream extends OutputStream
}
}
+ /**
+ * Waits for every queued part, then either finalizes the multipart upload
or aborts it. An abort discards the parts
+ * uploaded so far and leaves no object at {@link #s3Key} and finally throws
a {@link DruidException} to ensure
+ * callers know an error occurred and that the object is not available for
reading. The failure is an operator
+ * concern rather than a Druid defect: it means S3 rejected the parts.
+ */
private void completeMultipartUpload()
{
final List<CompletedPart> pushResults = new ArrayList<>();
+ Exception partUploadFailure = null;
for (Future<UploadPartResponse> future : futures) {
if (error) {
future.cancel(true);
@@ -311,6 +319,9 @@ public class RetryableS3OutputStream extends OutputStream
}
catch (Exception e) {
error = true;
+ if (partUploadFailure == null) {
+ partUploadFailure = e;
+ }
LOG.error(e, "Error in uploading part for upload ID [%s]", uploadId);
}
}
@@ -350,6 +361,20 @@ public class RetryableS3OutputStream extends OutputStream
catch (Exception e) {
throw new RuntimeException(e);
}
+
+ // If an error occurred during the upload of any part, we aborted the
whole upload and there is nothing to read
+ // downstream. We must throw here to indicate that the object was not
written and avoid callers assuming that the
+ // object is available for reading.
+ if (error) {
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+ .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+ .build(
+ partUploadFailure,
+ "Aborted multipart upload[%s] for s3Key[%s]; no
object was written",
+ uploadId,
+ s3Key
+ );
+ }
}
private static class Chunk implements Closeable
diff --git
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/output/RetryableS3OutputStreamTest.java
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/output/RetryableS3OutputStreamTest.java
index 48d943c2a7d..d3116dd9933 100644
---
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/output/RetryableS3OutputStreamTest.java
+++
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/output/RetryableS3OutputStreamTest.java
@@ -20,6 +20,7 @@
package org.apache.druid.storage.s3.output;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.HumanReadableBytes;
import org.apache.druid.java.util.common.IOE;
@@ -197,25 +198,37 @@ public class RetryableS3OutputStreamTest
s3.assertCompleted(chunkSize, Integer.BYTES * 25);
}
+ /**
+ * A part that fails every retry leaves the multipart upload aborted and no
object at the key, so close() must report
+ * it. Returning normally would tell the caller its bytes are readable back
when they no longer exist anywhere.
+ */
@Test
- public void testFailToUploadAfterRetries() throws IOException
+ public void testFailToUploadAfterRetries()
{
final TestAmazonS3 s3 = new TestAmazonS3(3);
ByteBuffer bb = ByteBuffer.allocate(Integer.BYTES);
- try (RetryableS3OutputStream out =
- new RetryableS3OutputStream(config, s3, path, s3UploadManager)) {
- for (int i = 0; i < 2; i++) {
+ final DruidException e = Assertions.assertThrows(DruidException.class, ()
-> {
+ try (RetryableS3OutputStream out =
+ new RetryableS3OutputStream(config, s3, path, s3UploadManager))
{
+ for (int i = 0; i < 2; i++) {
+ bb.clear();
+ bb.putInt(i);
+ out.write(bb.array());
+ }
+
bb.clear();
- bb.putInt(i);
+ bb.putInt(3);
out.write(bb.array());
}
-
- bb.clear();
- bb.putInt(3);
- out.write(bb.array());
- }
-
+ });
+
+ Assertions.assertTrue(e.getMessage().contains("no object was written"),
e.getMessage());
+ Assertions.assertTrue(e.getMessage().contains(path), e.getMessage());
+ Assertions.assertNotNull(e.getCause());
+ // An aborted upload means S3 rejected the parts, which an operator can
act on, rather than a Druid defect.
+ Assertions.assertEquals(DruidException.Persona.OPERATOR,
e.getTargetPersona());
+ Assertions.assertEquals(DruidException.Category.RUNTIME_FAILURE,
e.getCategory());
s3.assertCancelled();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
