This is an automated email from the ASF dual-hosted git repository.
gianm 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 5c65e195b08 fix: Restore respect for druid configs controlling s3
multipart upload configs (#19890)
5c65e195b08 is described below
commit 5c65e195b083f0e3910562393d242ad60ee76d5a
Author: Lucas Capistrant <[email protected]>
AuthorDate: Wed Aug 5 23:07:06 2026 -0500
fix: Restore respect for druid configs controlling s3 multipart upload
configs (#19890)
---
docs/development/extensions-core/s3.md | 2 +-
.../apache/druid/storage/s3/S3StorageConfig.java | 3 +
.../apache/druid/storage/s3/S3TransferConfig.java | 15 +-
.../storage/s3/ServerSideEncryptingAmazonS3.java | 10 +-
.../druid/storage/s3/S3TransferConfigTest.java | 33 ++++
...rSideEncryptingAmazonS3MultipartUploadTest.java | 181 +++++++++++++++++++++
6 files changed, 240 insertions(+), 4 deletions(-)
diff --git a/docs/development/extensions-core/s3.md
b/docs/development/extensions-core/s3.md
index 32f09f93ff7..c59f12670fd 100644
--- a/docs/development/extensions-core/s3.md
+++ b/docs/development/extensions-core/s3.md
@@ -55,7 +55,7 @@ To use S3 for Deep Storage, you must supply [connection
information](#configurat
|`druid.storage.disableAcl`|Boolean flag for how object permissions are
handled. To use ACLs, set this property to `false`. To use Object Ownership,
set it to `true`. The permission requirements for ACLs and Object Ownership are
different. For more information, see [S3 permissions
settings](#s3-permissions-settings).|false|
|`druid.storage.zip`|`true`, `false`|Whether segments in `s3` are written as
directories (`false`) or zip files (`true`).|`false`|
|`druid.storage.transfer.useTransferManager`| If true, use AWS S3 Transfer
Manager to upload segments to S3.|true|
-|`druid.storage.transfer.minimumUploadPartSize`| Minimum size (in bytes) of
each part in a multipart upload.|20971520 (20 MB)|
+|`druid.storage.transfer.minimumUploadPartSize`| Minimum size (in bytes) of
each part in a multipart upload. Must be between 5242880 (5 MiB) and 5368709120
(5 GiB), the range S3 permits for every part except the last. Larger parts mean
fewer requests per upload, which matters when many tasks write concurrently
under one key prefix.|20971520 (20 MB)|
|`druid.storage.transfer.multipartUploadThreshold`| The file size threshold
(in bytes) above which a file upload is converted into a multipart upload
instead of a single PUT request.| 20971520 (20 MB)|
|`druid.storage.transfer.asyncHttpClientType`| Async HTTP client
implementation used by the S3 Transfer Manager. Accepted values: `crt` (Amazon
CRT) or `netty` (Netty NIO).|`crt`|
diff --git
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3StorageConfig.java
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3StorageConfig.java
index b52d13cd518..6ebd08f8bd3 100644
---
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3StorageConfig.java
+++
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3StorageConfig.java
@@ -22,6 +22,8 @@ package org.apache.druid.storage.s3;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.Valid;
+
/**
* General configurations for Amazon S3 storage.
*/
@@ -42,6 +44,7 @@ public class S3StorageConfig
* @see S3StorageDruidModule#configure
*/
@JsonProperty("transfer")
+ @Valid
private final S3TransferConfig s3TransferConfig;
@JsonCreator
diff --git
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3TransferConfig.java
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3TransferConfig.java
index 5cd13e20b4e..655ac466877 100644
---
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3TransferConfig.java
+++
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3TransferConfig.java
@@ -20,7 +20,9 @@
package org.apache.druid.storage.s3;
import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.storage.s3.output.S3OutputConfig;
+import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
/**
@@ -30,10 +32,21 @@ public class S3TransferConfig
@JsonProperty
private boolean useTransferManager = true;
+ /**
+ * Size of each part of a multipart upload except the last, which S3
requires to be between 5MiB and 5GiB. A value
+ * outside that range is rejected at configuration time rather than
surfacing as an {@code EntityTooSmall} or
+ * {@code EntityTooLarge} on every upload.
+ */
@JsonProperty
- @Min(1)
+ @Min(S3OutputConfig.S3_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES)
+ @Max(S3OutputConfig.S3_MULTIPART_UPLOAD_MAX_PART_SIZE_BYTES)
private long minimumUploadPartSize = 20 * 1024 * 1024L;
+ /**
+ * Upload size at or above which multipart is used instead of a single PUT.
Unlike {@link #minimumUploadPartSize}
+ * this has no 5MiB floor: S3 exempts the final part of an upload from the
minimum, so an upload just over a small
+ * threshold is still valid as a single undersized part.
+ */
@JsonProperty
@Min(1)
private long multipartUploadThreshold = 20 * 1024 * 1024L;
diff --git
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3.java
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3.java
index b5b926ec71c..dd86e9e610d 100644
---
a/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3.java
+++
b/extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3.java
@@ -71,6 +71,7 @@ import software.amazon.awssdk.services.s3.model.S3Exception;
import software.amazon.awssdk.services.s3.model.Type;
import software.amazon.awssdk.services.s3.model.UploadPartRequest;
import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
import software.amazon.awssdk.services.sts.StsClient;
import software.amazon.awssdk.services.sts.StsClientBuilder;
import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
@@ -376,10 +377,15 @@ public class ServerSideEncryptingAmazonS3
clientBuilder.serviceConfiguration(s3Config)
.forcePathStyle(awsClientConfig.isEnablePathStyleAccess())
.crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled());
+ final S3TransferConfig transferConfig =
s3StorageConfig.getS3TransferConfig();
asyncClientBuilder.forcePathStyle(awsClientConfig.isEnablePathStyleAccess())
.crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled())
-
.httpClientBuilder(AsyncHttpClientType.fromString(s3StorageConfig.getS3TransferConfig().getAsyncHttpClientType()).buildBuilder(awsClientConfig))
- .multipartEnabled(true);
+
.httpClientBuilder(AsyncHttpClientType.fromString(transferConfig.getAsyncHttpClientType()).buildBuilder(awsClientConfig))
+ .multipartEnabled(true)
+ .multipartConfiguration(
+
MultipartConfiguration.builder().minimumPartSizeInBytes(transferConfig.getMinimumUploadPartSize())
+
.thresholdInBytes(transferConfig.getMultipartUploadThreshold())
+ .build());
}
// Configure HTTP client with proxy if needed
diff --git
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TransferConfigTest.java
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TransferConfigTest.java
index 080051c180d..7379fc4fbfa 100644
---
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TransferConfigTest.java
+++
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3TransferConfigTest.java
@@ -19,11 +19,44 @@
package org.apache.druid.storage.s3;
+import org.apache.druid.storage.s3.output.S3OutputConfig;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import javax.validation.Validation;
+import javax.validation.Validator;
+
public class S3TransferConfigTest
{
+ private static final Validator VALIDATOR =
Validation.buildDefaultValidatorFactory().getValidator();
+
+ @Test
+ public void testPartSizeBelowTheS3MinimumIsRejected()
+ {
+ final S3TransferConfig config = new S3TransferConfig();
+
config.setMinimumUploadPartSize(S3OutputConfig.S3_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES
- 1);
+
+ Assertions.assertFalse(VALIDATOR.validate(new S3StorageConfig(null,
config)).isEmpty());
+ }
+
+ @Test
+ public void testPartSizeAboveTheS3MaximumIsRejected()
+ {
+ final S3TransferConfig config = new S3TransferConfig();
+
config.setMinimumUploadPartSize(S3OutputConfig.S3_MULTIPART_UPLOAD_MAX_PART_SIZE_BYTES
+ 1);
+
+ Assertions.assertFalse(VALIDATOR.validate(new S3StorageConfig(null,
config)).isEmpty());
+ }
+
+ @Test
+ public void testThresholdBelowTheS3MinimumPartSizeIsAccepted()
+ {
+ final S3TransferConfig config = new S3TransferConfig();
+ config.setMultipartUploadThreshold(1024L);
+
+ Assertions.assertTrue(VALIDATOR.validate(new S3StorageConfig(null,
config)).isEmpty());
+ }
+
@Test
public void testDefaultValues()
{
diff --git
a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3MultipartUploadTest.java
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3MultipartUploadTest.java
new file mode 100644
index 00000000000..c13d1f59f7a
--- /dev/null
+++
b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3MultipartUploadTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.druid.storage.s3;
+
+import org.apache.druid.common.aws.AWSClientConfig;
+import org.apache.druid.common.aws.AWSEndpointConfig;
+import org.apache.druid.java.util.common.StringUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.testcontainers.containers.MinIOContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.services.s3.model.CreateBucketRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+
+/**
+ * Verifies that {@code druid.storage.transfer.minimumUploadPartSize} and
+ * {@code druid.storage.transfer.multipartUploadThreshold} govern how the S3
client splits an upload into parts.
+ *
+ * <p>Part size determines how many PUT-class requests a single upload costs,
which is what the per-prefix S3
+ * request-rate budget is spent on. The client exercised here is the one
{@link S3StorageDruidModule} provides and
+ * that {@link S3DataSegmentPusher} pushes segments through, so the part size
observed here is the part size an
+ * indexing task pays for every segment it writes.
+ *
+ * <p>The built {@code S3AsyncClient} keeps its multipart settings in
SDK-internal fields, so these tests assert on
+ * what S3 actually received: a multipart ETag carries a {@code -<partCount>}
suffix, a single-PUT ETag does not.
+ */
+@Testcontainers
+@Tag("requires-dockerd")
+public class ServerSideEncryptingAmazonS3MultipartUploadTest
+{
+ private static final String BUCKET = "testbucket";
+ private static final long MIB = 1024 * 1024L;
+
+ /**
+ * Deliberately different from the AWS SDK's own 8 MiB default, so a passing
test means the configured value was
+ * applied rather than a default that happens to be close.
+ */
+ private static final long CONFIGURED_PART_SIZE = 16 * MIB;
+
+ @Container
+ private static final MinIOContainer MINIO =
+ new
MinIOContainer(DockerImageName.parse("minio/minio:latest")).withEnv("MINIO_DOMAIN",
"localhost");
+
+ @TempDir
+ public File temporaryFolder;
+
+ private ServerSideEncryptingAmazonS3 s3;
+
+ @BeforeEach
+ public void setUp()
+ {
+ final S3TransferConfig transferConfig = new S3TransferConfig();
+ transferConfig.setUseTransferManager(true);
+ transferConfig.setMinimumUploadPartSize(CONFIGURED_PART_SIZE);
+ transferConfig.setMultipartUploadThreshold(CONFIGURED_PART_SIZE);
+
+ final AWSEndpointConfig endpointConfig = new AWSEndpointConfig()
+ {
+ @Override
+ public String getUrl()
+ {
+ return MINIO.getS3URL();
+ }
+
+ @Override
+ public String getSigningRegion()
+ {
+ return "us-east-1";
+ }
+ };
+
+ // MinIO is reached by host:port, so bucket-as-subdomain addressing will
not resolve.
+ final AWSClientConfig clientConfig = new AWSClientConfig()
+ {
+ @Override
+ public boolean isEnablePathStyleAccess()
+ {
+ return true;
+ }
+ };
+
+ s3 = ServerSideEncryptingAmazonS3.builder(
+
StaticCredentialsProvider.create(AwsBasicCredentials.create(MINIO.getUserName(),
MINIO.getPassword())),
+ new S3StorageConfig(new NoopServerSideEncryption(), transferConfig),
+ null,
+ endpointConfig,
+ clientConfig,
+ null,
+ null
+ ).build();
+
+ try {
+ s3.getS3Client().headBucket(b -> b.bucket(BUCKET));
+ }
+ catch (S3Exception e) {
+ if (e.statusCode() == 404) {
+
s3.getS3Client().createBucket(CreateBucketRequest.builder().bucket(BUCKET).build());
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ @Test
+ public void testUploadSplitsFileIntoPartsOfTheConfiguredSize() throws
IOException
+ {
+ final long fileSize = 3 * CONFIGURED_PART_SIZE;
+ final String key = "part-size/upload.bin";
+
+ s3.upload(BUCKET, key, fileOfSize("upload.bin", fileSize), null);
+
+ Assertions.assertEquals(
+ 3,
+ partCountOf(S3Utils.getSingleObjectMetadata(s3, BUCKET, key).eTag()),
+ "a file of exactly three configured parts should be uploaded as three
parts"
+ );
+ }
+
+ @Test
+ public void testUploadOfFileBelowConfiguredThresholdUsesASinglePut() throws
IOException
+ {
+ final long fileSize = CONFIGURED_PART_SIZE - (4 * MIB);
+ final String key = "threshold/upload.bin";
+
+ s3.upload(BUCKET, key, fileOfSize("small.bin", fileSize), null);
+
+ Assertions.assertEquals(
+ 1,
+ partCountOf(S3Utils.getSingleObjectMetadata(s3, BUCKET, key).eTag()),
+ "a file under the configured multipart threshold should not be split
at all"
+ );
+ }
+
+ private File fileOfSize(String name, long size) throws IOException
+ {
+ final File file = new File(temporaryFolder, name);
+ try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
+ raf.setLength(size);
+ }
+ return file;
+ }
+
+ /**
+ * S3 reports a multipart object's ETag as {@code <hash>-<partCount>}; an
object written with a single PUT has no
+ * suffix.
+ */
+ private static int partCountOf(String eTag)
+ {
+ final String unquoted = StringUtils.replace(eTag, "\"", "");
+ final int dash = unquoted.lastIndexOf('-');
+ return dash < 0 ? 1 : Integer.parseInt(unquoted.substring(dash + 1));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
