From 43c5ade447526c929111cedfe8abaeddb408f2f0 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 5 Aug 2026 15:14:56 -0500 Subject: [PATCH 01/10] Add operator access to configure AWS retry policy. Default to STANDARD, recommended by AWS --- cloud/aws-common/pom.xml | 11 ++ .../druid/common/aws/AWSClientConfig.java | 129 ++++++++++++++++++ .../druid/common/aws/AWSClientConfigTest.java | 82 +++++++++++ docs/development/extensions-core/s3.md | 57 ++++++++ .../s3/ServerSideEncryptingAmazonS3.java | 22 ++- 5 files changed, 297 insertions(+), 4 deletions(-) diff --git a/cloud/aws-common/pom.xml b/cloud/aws-common/pom.xml index d1197a934d58..d27ae8ffe8e1 100644 --- a/cloud/aws-common/pom.xml +++ b/cloud/aws-common/pom.xml @@ -82,6 +82,17 @@ sdk-core ${aws.sdk.v2.version} + + + software.amazon.awssdk + retries-spi + ${aws.sdk.v2.version} + + + software.amazon.awssdk + retries + ${aws.sdk.v2.version} + software.amazon.awssdk diff --git a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java index 9fdf9ba592f7..858fbf8a345a 100644 --- a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java +++ b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java @@ -20,10 +20,18 @@ package org.apache.druid.common.aws; import com.fasterxml.jackson.annotation.JacksonInject; +import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonValue; +import org.apache.druid.java.util.common.IAE; +import org.apache.druid.java.util.common.StringUtils; import org.apache.druid.utils.RuntimeInfo; +import software.amazon.awssdk.awscore.retry.AwsRetryStrategy; +import software.amazon.awssdk.retries.api.RetryStrategy; import javax.annotation.Nullable; +import javax.validation.constraints.Min; +import java.util.Arrays; public class AWSClientConfig { @@ -36,6 +44,68 @@ public class AWSClientConfig /** AWS SDK v2's own default. */ private static final int DEFAULT_MAX_CONNECTIONS_FLOOR = 50; + /** + * Selects the retry behaviour AWS documents for {@code standard} and {@code adaptive} rather than the pre-2026 + * behaviour. The no-argument factories default the {@code aws.newRetries2026} opt-in to false, which changes the + * backoff base delays and the retry-quota token costs; asking for it explicitly means the modes behave as + * documented, and keeps them behaving that way when AWS flips the opt-in default. + */ + private static final Boolean USE_DOCUMENTED_RETRY_BEHAVIOUR = Boolean.TRUE; + + /** + * Retry strategy family. Declared as an enum so an unrecognised value is rejected while the config is bound at + * startup, rather than when a client is first built. + */ + public enum RetryMode + { + STANDARD { + @Override + RetryStrategy createStrategy() + { + return AwsRetryStrategy.standardRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOUR); + } + }, + ADAPTIVE { + @Override + RetryStrategy createStrategy() + { + // Standard plus a client-side rate limiter, which unlike standard can delay or block the initial request, + // not just retries. The limiter belongs to one client instance and covers every request that client makes, + // so throttling on one key prefix also slows requests to prefixes that are not being throttled. + return AwsRetryStrategy.adaptiveRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOUR); + } + }, + LEGACY { + @Override + RetryStrategy createStrategy() + { + // Deliberately left on the pre-standard behaviour: this mode exists so a deployment can get back to what it + // had before, which is the opposite of what the opt-in above asks for. + return AwsRetryStrategy.legacyRetryStrategy(); + } + }; + + abstract RetryStrategy createStrategy(); + + @JsonValue + @Override + public String toString() + { + return StringUtils.toLowerCase(name()); + } + + @JsonCreator + public static RetryMode fromString(String value) + { + for (RetryMode mode : values()) { + if (mode.name().equalsIgnoreCase(value)) { + return mode; + } + } + throw new IAE("Invalid druid.s3.retryMode[%s]. Must be one of %s.", value, Arrays.toString(values())); + } + } + /** * Used by {@link #getMaxConnections} to scale the default connection pool with host size so hosts large enough to * do a lot of concurrent deep-storage I/O (e.g. virtual-storage historicals fanning out on-demand loads to S3) @@ -80,6 +150,30 @@ public class AWSClientConfig @Nullable private Integer maxConnections = null; + /** + * Retry strategy applied to every AWS client built from this config. + *

+ * Setting this at all is deliberate: left unset, the SDK picks its own default, and which one it picks depends on + * the {@code aws.newRetries2026} migration flag. Naming the mode here keeps retry behaviour stable across SDK + * upgrades instead of changing under Druid when that flag's default flips. + */ + @JsonProperty + private RetryMode retryMode = RetryMode.STANDARD; + + /** + * Total attempts per request, including the first. Maps directly to the SDK's {@code maxAttempts}, so 1 disables + * retries. Null leaves the count that {@link #retryMode} defines for itself, which AWS tunes alongside that mode's + * backoff and retry quota. + *

+ * This counts HTTP requests. Druid layers its own retries on top (see {@code S3Utils#retryS3Operation}) and the two + * multiply, but they are not equivalent: an attempt here re-sends a single request, whereas a Druid-level retry + * repeats a whole operation, such as re-uploading an entire segment. + */ + @JsonProperty + @Nullable + @Min(1) + private Integer maxRetryAttempts = null; + public String getProtocol() { return protocol; @@ -146,6 +240,39 @@ public int getMaxConnections() return Math.max(DEFAULT_MAX_CONNECTIONS_FLOOR, 4 * runtimeInfo.getAvailableProcessors()); } + public RetryMode getRetryMode() + { + return retryMode; + } + + @Nullable + public Integer getMaxRetryAttempts() + { + return maxRetryAttempts; + } + + /** + * Builds the strategy to hand to {@code ClientOverrideConfiguration.retryStrategy}. Kept as a plain function of the + * config because a built AWS client does not expose the strategy it was given, so this is the only place the + * mapping can be tested. + */ + public RetryStrategy getRetryStrategy() + { + return withMaxAttempts(retryMode.createStrategy()); + } + + /** + * Overrides the attempt count only when one is configured, so an unset {@link #maxRetryAttempts} leaves whatever + * the chosen mode defines for itself. + */ + private RetryStrategy withMaxAttempts(RetryStrategy strategy) + { + if (maxRetryAttempts == null) { + return strategy; + } + return strategy.toBuilder().maxAttempts(maxRetryAttempts).build(); + } + @Override public String toString() { @@ -157,6 +284,8 @@ public String toString() ", connectionTimeout=" + connectionTimeout + ", socketTimeout=" + socketTimeout + ", maxConnections=" + getMaxConnections() + + ", retryMode='" + retryMode + '\'' + + ", maxRetryAttempts=" + maxRetryAttempts + '}'; } } diff --git a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java index 99b927efb2ba..bd91b5898090 100644 --- a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java +++ b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java @@ -20,10 +20,16 @@ package org.apache.druid.common.aws; import com.fasterxml.jackson.databind.InjectableValues; +import com.fasterxml.jackson.databind.JsonMappingException; import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.druid.java.util.common.IAE; import org.apache.druid.utils.RuntimeInfo; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import software.amazon.awssdk.awscore.retry.AwsRetryStrategy; +import software.amazon.awssdk.retries.AdaptiveRetryStrategy; +import software.amazon.awssdk.retries.LegacyRetryStrategy; +import software.amazon.awssdk.retries.StandardRetryStrategy; public class AWSClientConfigTest { @@ -38,6 +44,82 @@ private static ObjectMapper mapperWithRuntimeInfo(RuntimeInfo runtimeInfo) ); } + @Test + public void testDefaultRetryModeIsStandard() + { + final AWSClientConfig config = new AWSClientConfig(); + + Assertions.assertInstanceOf(StandardRetryStrategy.class, config.getRetryStrategy()); + } + + @Test + public void testDefaultLeavesAttemptCountToTheSdk() + { + final AWSClientConfig config = new AWSClientConfig(); + + Assertions.assertNull(config.getMaxRetryAttempts()); + Assertions.assertEquals( + AwsRetryStrategy.standardRetryStrategy().maxAttempts(), + config.getRetryStrategy().maxAttempts() + ); + } + + @Test + public void testAdaptiveRetryModeIsSelectable() throws Exception + { + final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"adaptive\"}", AWSClientConfig.class); + + Assertions.assertInstanceOf(AdaptiveRetryStrategy.class, config.getRetryStrategy()); + } + + @Test + public void testLegacyRetryModeIsSelectable() throws Exception + { + final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"legacy\"}", AWSClientConfig.class); + + Assertions.assertInstanceOf(LegacyRetryStrategy.class, config.getRetryStrategy()); + } + + @Test + public void testRetryModeIsCaseInsensitive() throws Exception + { + final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"ADAPTIVE\"}", AWSClientConfig.class); + + Assertions.assertInstanceOf(AdaptiveRetryStrategy.class, config.getRetryStrategy()); + } + + @Test + public void testConfiguredAttemptCountIsApplied() throws Exception + { + final AWSClientConfig config = MAPPER.readValue("{\"maxRetryAttempts\": 8}", AWSClientConfig.class); + + Assertions.assertEquals(8, config.getRetryStrategy().maxAttempts()); + } + + /** + * Binding the config is the last point at which a bad mode can be reported against the property that set it, so it + * has to fail here rather than when some client is first built. + */ + @Test + public void testUnrecognizedRetryModeIsRejectedWhenConfigIsBound() + { + final JsonMappingException e = Assertions.assertThrows( + JsonMappingException.class, + () -> MAPPER.readValue("{\"retryMode\": \"aggressive\"}", AWSClientConfig.class) + ); + + Assertions.assertInstanceOf(IAE.class, e.getCause()); + Assertions.assertTrue(e.getCause().getMessage().contains("aggressive")); + } + + @Test + public void testRetryModeSerializesToItsPropertyValue() throws Exception + { + final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"adaptive\"}", AWSClientConfig.class); + + Assertions.assertEquals("adaptive", MAPPER.convertValue(config.getRetryMode(), String.class)); + } + @Test public void testDefaultCrossRegionAccessEnabled() throws Exception { diff --git a/docs/development/extensions-core/s3.md b/docs/development/extensions-core/s3.md index 32f09f93ff7e..c6dd821182ad 100644 --- a/docs/development/extensions-core/s3.md +++ b/docs/development/extensions-core/s3.md @@ -131,6 +131,8 @@ For example, to set the region to 'us-east-1' through system properties: |`druid.s3.disableChunkedEncoding`|Disables chunked encoding. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#disableChunkedEncoding--) for details.|false| |`druid.s3.enablePathStyleAccess`|Enables path style access. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#enablePathStyleAccess--) for details.|false| |`druid.s3.crossRegionAccessEnabled`|Enables cross-region access for S3 requests. When enabled, the S3 client automatically detects the correct region for a bucket on first access and caches it for subsequent requests.|false| +|`druid.s3.retryMode`|Retry strategy used by every AWS client Druid builds. One of `standard`, `adaptive` or `legacy`. See [Retry behavior](#retry-behavior).|`standard`| +|`druid.s3.maxRetryAttempts`|Total attempts per HTTP request, including the first, so `1` disables SDK retries. When unset, the count defined by `druid.s3.retryMode` applies. See [Retry behavior](#retry-behavior).|null (the retry mode's own default)| |`druid.s3.forceGlobalBucketAccessEnabled`|**Deprecated.** Use `druid.s3.crossRegionAccessEnabled` instead. Only used as a fallback if `crossRegionAccessEnabled` is not explicitly set.|null| |`druid.s3.endpoint.url`|Service endpoint either with or without the protocol.|None| |`druid.s3.endpoint.signingRegion`|Region to use for SigV4 signing of requests (e.g. us-west-1).|None| @@ -142,6 +144,61 @@ For example, to set the region to 'us-east-1' through system properties: |`druid.storage.sse.kms.keyId`|AWS KMS key ID. This is used only when `druid.storage.sse.type` is `kms` and can be empty to use the default key ID.|None| |`druid.storage.sse.custom.base64EncodedKey`|Base64-encoded key. Should be specified if `druid.storage.sse.type` is `custom`.|None| +## Retry behavior + +Druid names the AWS SDK retry strategy explicitly rather than accepting the SDK's default. The SDK's own default +depends on the `aws.newRetries2026` migration flag, so leaving it unset would let retry behavior change underneath +Druid when that flag's default flips. For `standard` and `adaptive`, Druid also opts in to the behavior AWS +documents for those modes, rather than quietly falling back to any legacy behavior the SDK may have. + +|Mode|Behavior| +|----|--------| +|`standard`|Error classification consistent with the other AWS SDK implementations, and the mode AWS recommends for all workloads.| +|`adaptive`|`standard` plus a client-side rate limiter that slows requests down when S3 reports throttling. Unlike `standard`, it can delay or block the **initial** request, not only retries. The limiter covers every request made by one client instance, so throttling on one key prefix also slows requests to prefixes that are not being throttled.| +|`legacy`|The SDK's legacy behavior, retained so a deployment can revert without a rollback. AWS recommends moving off it.| + +### Choosing a mode + +AWS recommends `standard` as the default and `adaptive` only for workloads that are *single-resource, +throttling-heavy, and latency-tolerant*. S3 applies its request-rate limits per key prefix, so "single-resource" +means a process that concentrates its requests on one prefix. + +That maps onto Druid roughly as follows. Because a peon runs one task, you can select a mode per task type — through +a Kubernetes pod template, or `druid.indexer.fork.property.druid.s3.retryMode` in a task's context when using the +MiddleManager task runner. + +|Process|Suggested mode|Reasoning| +|-------|--------------|---------| +|MSQ and compaction peons|`adaptive`|Segment output goes to one new version prefix and shuffle output to one prefix per query, which is exactly the concentration that provokes throttling. Batch work tolerates the added latency.| +|Batch ingestion peons|`standard`, or `adaptive` if throttled|Same shape as above at lower request rates.| +|Historicals|`standard`|One client loads segments for every datasource the process serves, so a rate limiter tripped by one prefix would slow loads for unrelated ones. Segment loads can also sit on the query path.| +|Brokers, Coordinator, Overlord|`standard`|Low request volume, and delaying an initial request costs query latency for no benefit.| + +### Retry quota + +Every mode carries a retry quota: a token bucket, held per client instance and never shared across processes, that +stops retries once it is exhausted so the client fails fast instead of adding load a struggling service cannot +absorb. It only engages under sustained failure — roughly a 32% failure rate for throttling errors — and is inert +otherwise. Choosing between the modes does not change whether it is present. + +S3 reports throttling as `SlowDown` and `503`, which the SDK classifies as throttling rather than transient errors. +Those get a longer base backoff than transient failures, and they are what `adaptive`'s rate limiter reacts to. + +### Retries are layered + +`druid.s3.maxRetryAttempts` applies per HTTP request. Druid retries again on top of it, and the two multiply: + +|Layer|Scope of one attempt|Attempts|Backoff cap| +|-----|--------------------|--------|-----------| +|`druid.s3.maxRetryAttempts`|A single HTTP request, such as one `UploadPart`|Set by the retry mode|`~20s`| +|`S3Utils.retryS3Operation`|A whole logical operation, such as re-uploading an entire segment|10|`60s`| +|`druid.msq.intermediate.storage.maxRetry`|A durable storage operation|10|`60s`| + +The layers are not equivalent, and the difference matters more than the totals: an attempt at the first layer +re-sends one part, whereas a retry at the layers below it repeats the whole transfer. A workload that keeps losing +uploads to throttling is usually better served by raising `druid.s3.maxRetryAttempts`, which absorbs failures before +they reach a layer that re-uploads everything, than by raising the Druid-level counts. + ## Server-side encryption You can enable [server-side encryption](https://docs.aws.amazon.com/AmazonS3/latest/dev/serv-side-encryption) by setting 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 b5b926ec71ce..9bdbacabdca8 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 @@ -35,6 +35,7 @@ import software.amazon.awssdk.auth.credentials.AwsSessionCredentials; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.core.ResponseInputStream; +import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration; import software.amazon.awssdk.core.exception.SdkClientException; import software.amazon.awssdk.core.sync.RequestBody; import software.amazon.awssdk.http.apache.ApacheHttpClient; @@ -373,10 +374,16 @@ public static ServerSideEncryptingAmazonS3.Builder builder( S3Configuration s3Config = S3Configuration.builder() .chunkedEncodingEnabled(!awsClientConfig.isDisableChunkedEncoding()) .build(); + final ClientOverrideConfiguration retryOverrides = + ClientOverrideConfiguration.builder() + .retryStrategy(awsClientConfig.getRetryStrategy()) + .build(); clientBuilder.serviceConfiguration(s3Config) .forcePathStyle(awsClientConfig.isEnablePathStyleAccess()) - .crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled()); - asyncClientBuilder.forcePathStyle(awsClientConfig.isEnablePathStyleAccess()) + .crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled()) + .overrideConfiguration(retryOverrides); + asyncClientBuilder.overrideConfiguration(retryOverrides) + .forcePathStyle(awsClientConfig.isEnablePathStyleAccess()) .crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled()) .httpClientBuilder(AsyncHttpClientType.fromString(s3StorageConfig.getS3TransferConfig().getAsyncHttpClientType()).buildBuilder(awsClientConfig)) .multipartEnabled(true); @@ -405,7 +412,8 @@ public static ServerSideEncryptingAmazonS3.Builder builder( assumeRoleArn, assumeRoleExternalId, awsEndpointConfig, - credentialsProvider + credentialsProvider, + awsClientConfig ); } @@ -443,7 +451,8 @@ public static AwsCredentialsProvider createAssumeRoleCredentialsProvider( String assumeRoleArn, @Nullable String assumeRoleExternalId, @Nullable AWSEndpointConfig awsEndpointConfig, - AwsCredentialsProvider baseCredentialsProvider + AwsCredentialsProvider baseCredentialsProvider, + @Nullable AWSClientConfig awsClientConfig ) { String roleSessionName = StringUtils.format("druid-s3-%s", UUID.randomUUID().toString()); @@ -453,6 +462,11 @@ public static AwsCredentialsProvider createAssumeRoleCredentialsProvider( if (awsEndpointConfig != null && awsEndpointConfig.getSigningRegion() != null) { stsBuilder.region(Region.of(awsEndpointConfig.getSigningRegion())); } + if (awsClientConfig != null) { + stsBuilder.overrideConfiguration( + ClientOverrideConfiguration.builder().retryStrategy(awsClientConfig.getRetryStrategy()).build() + ); + } AssumeRoleRequest.Builder assumeRoleRequestBuilder = AssumeRoleRequest.builder().roleArn(assumeRoleArn).roleSessionName(roleSessionName).durationSeconds(3600); From 4dcd46f22501ef4fbe1b6a3d0ba2c7ca90ae7003 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 5 Aug 2026 16:33:12 -0500 Subject: [PATCH 02/10] add missing dep --- extensions-core/s3-extensions/pom.xml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/extensions-core/s3-extensions/pom.xml b/extensions-core/s3-extensions/pom.xml index 40b3b549fd51..9b6581e84345 100644 --- a/extensions-core/s3-extensions/pom.xml +++ b/extensions-core/s3-extensions/pom.xml @@ -163,6 +163,11 @@ utils ${aws.sdk.v2.version} + + software.amazon.awssdk + retries-spi + ${aws.sdk.v2.version} + it.unimi.dsi fastutil-core From 6154897a77666accb8d863551d17f1cc3d3269d9 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 5 Aug 2026 16:47:15 -0500 Subject: [PATCH 03/10] modify some test structure --- cloud/aws-common/pom.xml | 5 + .../druid/common/aws/AWSClientConfigTest.java | 197 +++++++++--------- 2 files changed, 106 insertions(+), 96 deletions(-) diff --git a/cloud/aws-common/pom.xml b/cloud/aws-common/pom.xml index d27ae8ffe8e1..ab6646799244 100644 --- a/cloud/aws-common/pom.xml +++ b/cloud/aws-common/pom.xml @@ -126,5 +126,10 @@ junit-jupiter-api test + + org.junit.jupiter + junit-jupiter-params + test + diff --git a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java index bd91b5898090..8058405f9020 100644 --- a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java +++ b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java @@ -20,22 +20,43 @@ package org.apache.druid.common.aws; import com.fasterxml.jackson.databind.InjectableValues; -import com.fasterxml.jackson.databind.JsonMappingException; import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; import org.apache.druid.java.util.common.IAE; import org.apache.druid.utils.RuntimeInfo; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; -import software.amazon.awssdk.awscore.retry.AwsRetryStrategy; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; import software.amazon.awssdk.retries.AdaptiveRetryStrategy; import software.amazon.awssdk.retries.LegacyRetryStrategy; import software.amazon.awssdk.retries.StandardRetryStrategy; +import software.amazon.awssdk.retries.api.RetryStrategy; + +import java.util.Map; +import java.util.stream.Stream; public class AWSClientConfigTest { - private static final ObjectMapper MAPPER = new ObjectMapper().setInjectableValues( - new InjectableValues.Std().addValue(RuntimeInfo.class, new RuntimeInfo()) - ); + private static final ObjectMapper MAPPER = mapperWithRuntimeInfo(new RuntimeInfo()); + + /** + * Binds a property map the way {@code JsonConfigurator} binds {@code druid.s3.*} at startup, so behaviour that only + * exists during binding - defaults, unset versus explicitly set, rejection of bad values - is exercised here the + * same way it happens in a running process. + */ + private static AWSClientConfig bind(Map properties) + { + return MAPPER.convertValue(properties, AWSClientConfig.class); + } + + private static AWSClientConfig bind(Map properties, RuntimeInfo runtimeInfo) + { + return mapperWithRuntimeInfo(runtimeInfo).convertValue(properties, AWSClientConfig.class); + } private static ObjectMapper mapperWithRuntimeInfo(RuntimeInfo runtimeInfo) { @@ -49,51 +70,58 @@ public void testDefaultRetryModeIsStandard() { final AWSClientConfig config = new AWSClientConfig(); + Assertions.assertEquals(AWSClientConfig.RetryMode.STANDARD, config.getRetryMode()); Assertions.assertInstanceOf(StandardRetryStrategy.class, config.getRetryStrategy()); } - @Test - public void testDefaultLeavesAttemptCountToTheSdk() + @ParameterizedTest(name = "{0}") + @MethodSource("retryModeStrategies") + public void testEachRetryModeBuildsItsStrategy( + AWSClientConfig.RetryMode mode, + Class expected + ) { - final AWSClientConfig config = new AWSClientConfig(); - - Assertions.assertNull(config.getMaxRetryAttempts()); - Assertions.assertEquals( - AwsRetryStrategy.standardRetryStrategy().maxAttempts(), - config.getRetryStrategy().maxAttempts() - ); + Assertions.assertInstanceOf(expected, mode.createStrategy()); } + /** + * Guards {@link #retryModeStrategies} against a mode being added without a strategy expectation. + */ @Test - public void testAdaptiveRetryModeIsSelectable() throws Exception + public void testEveryRetryModeHasAStrategyExpectation() { - final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"adaptive\"}", AWSClientConfig.class); - - Assertions.assertInstanceOf(AdaptiveRetryStrategy.class, config.getRetryStrategy()); + Assertions.assertEquals(AWSClientConfig.RetryMode.values().length, retryModeStrategies().count()); } - @Test - public void testLegacyRetryModeIsSelectable() throws Exception + private static Stream retryModeStrategies() { - final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"legacy\"}", AWSClientConfig.class); + return Stream.of( + Arguments.of(AWSClientConfig.RetryMode.STANDARD, StandardRetryStrategy.class), + Arguments.of(AWSClientConfig.RetryMode.ADAPTIVE, AdaptiveRetryStrategy.class), + Arguments.of(AWSClientConfig.RetryMode.LEGACY, LegacyRetryStrategy.class) + ); + } - Assertions.assertInstanceOf(LegacyRetryStrategy.class, config.getRetryStrategy()); + @ParameterizedTest + @ValueSource(strings = {"adaptive", "ADAPTIVE", "Adaptive"}) + public void testRetryModeParsingIsCaseInsensitive(String value) + { + Assertions.assertEquals(AWSClientConfig.RetryMode.ADAPTIVE, AWSClientConfig.RetryMode.fromString(value)); } @Test - public void testRetryModeIsCaseInsensitive() throws Exception + public void testRetryModeBindsFromItsProperty() { - final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"ADAPTIVE\"}", AWSClientConfig.class); - - Assertions.assertInstanceOf(AdaptiveRetryStrategy.class, config.getRetryStrategy()); + Assertions.assertEquals( + AWSClientConfig.RetryMode.ADAPTIVE, + bind(Map.of("retryMode", "adaptive")).getRetryMode() + ); } @Test - public void testConfiguredAttemptCountIsApplied() throws Exception + public void testRetryModeSerializesToItsPropertyValue() { - final AWSClientConfig config = MAPPER.readValue("{\"maxRetryAttempts\": 8}", AWSClientConfig.class); - - Assertions.assertEquals(8, config.getRetryStrategy().maxAttempts()); + Assertions.assertEquals("adaptive", MAPPER.convertValue(AWSClientConfig.RetryMode.ADAPTIVE, String.class)); } /** @@ -103,102 +131,79 @@ public void testConfiguredAttemptCountIsApplied() throws Exception @Test public void testUnrecognizedRetryModeIsRejectedWhenConfigIsBound() { - final JsonMappingException e = Assertions.assertThrows( - JsonMappingException.class, - () -> MAPPER.readValue("{\"retryMode\": \"aggressive\"}", AWSClientConfig.class) + final IllegalArgumentException e = Assertions.assertThrows( + IllegalArgumentException.class, + () -> bind(Map.of("retryMode", "aggressive")) ); - Assertions.assertInstanceOf(IAE.class, e.getCause()); - Assertions.assertTrue(e.getCause().getMessage().contains("aggressive")); + final Throwable rootCause = Throwables.getRootCause(e); + Assertions.assertInstanceOf(IAE.class, rootCause); + Assertions.assertTrue(rootCause.getMessage().contains("aggressive")); } @Test - public void testRetryModeSerializesToItsPropertyValue() throws Exception + public void testUnsetAttemptCountLeavesTheCountTheModeDefines() { - final AWSClientConfig config = MAPPER.readValue("{\"retryMode\": \"adaptive\"}", AWSClientConfig.class); - - Assertions.assertEquals("adaptive", MAPPER.convertValue(config.getRetryMode(), String.class)); - } - - @Test - public void testDefaultCrossRegionAccessEnabled() throws Exception - { - AWSClientConfig config = MAPPER.readValue("{}", AWSClientConfig.class); - Assertions.assertNull(config.isForceGlobalBucketAccessEnabled()); - Assertions.assertFalse(config.isCrossRegionAccessEnabled()); - } - - @Test - public void testCrossRegionAccessEnabledExplicitlySet() throws Exception - { - AWSClientConfig config = MAPPER.readValue("{\"crossRegionAccessEnabled\": true}", AWSClientConfig.class); - Assertions.assertNull(config.isForceGlobalBucketAccessEnabled()); - Assertions.assertTrue(config.isCrossRegionAccessEnabled()); - } + final AWSClientConfig config = new AWSClientConfig(); - @Test - public void testNewConfigTakesPrecedenceOverDeprecatedWhenBothSet() throws Exception - { - AWSClientConfig config = MAPPER.readValue( - "{\"forceGlobalBucketAccessEnabled\": true, \"crossRegionAccessEnabled\": false}", - AWSClientConfig.class + Assertions.assertNull(config.getMaxRetryAttempts()); + Assertions.assertEquals( + AWSClientConfig.RetryMode.STANDARD.createStrategy().maxAttempts(), + config.getRetryStrategy().maxAttempts() ); - Assertions.assertFalse(config.isCrossRegionAccessEnabled()); } @Test - public void testNewConfigTrueWinsOverDeprecatedFalse() throws Exception + public void testConfiguredAttemptCountIsApplied() { - AWSClientConfig config = MAPPER.readValue( - "{\"forceGlobalBucketAccessEnabled\": false, \"crossRegionAccessEnabled\": true}", - AWSClientConfig.class - ); - Assertions.assertTrue(config.isCrossRegionAccessEnabled()); + Assertions.assertEquals(8, bind(Map.of("maxRetryAttempts", 8)).getRetryStrategy().maxAttempts()); } - @Test - public void testDeprecatedForceGlobalBucketAccessAloneTrue() throws Exception + @ParameterizedTest(name = "{0}") + @MethodSource("crossRegionAccessBindings") + public void testCrossRegionAccessResolution(Map properties, boolean expected) { - AWSClientConfig config = MAPPER.readValue( - "{\"forceGlobalBucketAccessEnabled\": true}", - AWSClientConfig.class - ); - Assertions.assertTrue(config.isCrossRegionAccessEnabled()); + Assertions.assertEquals(expected, bind(properties).isCrossRegionAccessEnabled()); } - @Test - public void testDeprecatedNotSetFallsThroughToCrossRegion() throws Exception + private static Stream crossRegionAccessBindings() { - AWSClientConfig config = MAPPER.readValue( - "{\"crossRegionAccessEnabled\": true}", - AWSClientConfig.class + return Stream.of( + Arguments.of(Map.of(), false), + Arguments.of(Map.of("crossRegionAccessEnabled", true), true), + Arguments.of(Map.of("forceGlobalBucketAccessEnabled", true), true), + // the new property wins whichever way the two disagree + Arguments.of(Map.of("forceGlobalBucketAccessEnabled", true, "crossRegionAccessEnabled", false), false), + Arguments.of(Map.of("forceGlobalBucketAccessEnabled", false, "crossRegionAccessEnabled", true), true) ); - Assertions.assertNull(config.isForceGlobalBucketAccessEnabled()); - Assertions.assertTrue(config.isCrossRegionAccessEnabled()); } + /** + * The deprecated property is only ever populated by its own key, so code still reading it cannot be misled by the + * replacement being set. + */ @Test - public void testDefaultMaxConnectionsKeepsAwsSdkFloorOnSmallHost() throws Exception + @SuppressWarnings("deprecation") + public void testDeprecatedPropertyStaysUnsetWhenOnlyItsReplacementIsBound() { - AWSClientConfig config = mapperWithRuntimeInfo(new FixedProcessorsRuntimeInfo(8)) - .readValue("{}", AWSClientConfig.class); - Assertions.assertEquals(50, config.getMaxConnections()); + Assertions.assertNull(bind(Map.of()).isForceGlobalBucketAccessEnabled()); + Assertions.assertNull(bind(Map.of("crossRegionAccessEnabled", true)).isForceGlobalBucketAccessEnabled()); } - @Test - public void testDefaultMaxConnectionsScalesWithCoresOnLargeHost() throws Exception + @ParameterizedTest(name = "{0} processors -> {1} connections") + @CsvSource({"8, 50", "32, 128"}) + public void testDefaultMaxConnectionsTakesTheSdkFloorOrFourPerCore(int processors, int expected) { - AWSClientConfig config = mapperWithRuntimeInfo(new FixedProcessorsRuntimeInfo(32)) - .readValue("{}", AWSClientConfig.class); - Assertions.assertEquals(128, config.getMaxConnections()); + Assertions.assertEquals(expected, bind(Map.of(), new FixedProcessorsRuntimeInfo(processors)).getMaxConnections()); } @Test - public void testExplicitMaxConnectionsOverridesDefault() throws Exception + public void testExplicitMaxConnectionsOverridesDefault() { - AWSClientConfig config = mapperWithRuntimeInfo(new FixedProcessorsRuntimeInfo(64)) - .readValue("{\"maxConnections\": 200}", AWSClientConfig.class); - Assertions.assertEquals(200, config.getMaxConnections()); + Assertions.assertEquals( + 200, + bind(Map.of("maxConnections", 200), new FixedProcessorsRuntimeInfo(64)).getMaxConnections() + ); } private static final class FixedProcessorsRuntimeInfo extends RuntimeInfo From 7c7b3b899669ea287cb6b248ac6d663b01d4f22a Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 5 Aug 2026 17:04:13 -0500 Subject: [PATCH 04/10] remove robot affinity for british spelling --- .../apache/druid/common/aws/AWSClientConfig.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java index 858fbf8a345a..3379a00f994f 100644 --- a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java +++ b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java @@ -45,12 +45,12 @@ public class AWSClientConfig private static final int DEFAULT_MAX_CONNECTIONS_FLOOR = 50; /** - * Selects the retry behaviour AWS documents for {@code standard} and {@code adaptive} rather than the pre-2026 - * behaviour. The no-argument factories default the {@code aws.newRetries2026} opt-in to false, which changes the + * Selects the retry behavior AWS documents for {@code standard} and {@code adaptive} rather than the pre-2026 + * behavior. The no-argument factories default the {@code aws.newRetries2026} opt-in to false, which changes the * backoff base delays and the retry-quota token costs; asking for it explicitly means the modes behave as * documented, and keeps them behaving that way when AWS flips the opt-in default. */ - private static final Boolean USE_DOCUMENTED_RETRY_BEHAVIOUR = Boolean.TRUE; + private static final Boolean USE_DOCUMENTED_RETRY_BEHAVIOR = Boolean.TRUE; /** * Retry strategy family. Declared as an enum so an unrecognised value is rejected while the config is bound at @@ -62,7 +62,7 @@ public enum RetryMode @Override RetryStrategy createStrategy() { - return AwsRetryStrategy.standardRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOUR); + return AwsRetryStrategy.standardRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOR); } }, ADAPTIVE { @@ -72,14 +72,14 @@ RetryStrategy createStrategy() // Standard plus a client-side rate limiter, which unlike standard can delay or block the initial request, // not just retries. The limiter belongs to one client instance and covers every request that client makes, // so throttling on one key prefix also slows requests to prefixes that are not being throttled. - return AwsRetryStrategy.adaptiveRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOUR); + return AwsRetryStrategy.adaptiveRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOR); } }, LEGACY { @Override RetryStrategy createStrategy() { - // Deliberately left on the pre-standard behaviour: this mode exists so a deployment can get back to what it + // Deliberately left on the pre-standard behavior: this mode exists so a deployment can get back to what it // had before, which is the opposite of what the opt-in above asks for. return AwsRetryStrategy.legacyRetryStrategy(); } @@ -154,7 +154,7 @@ public static RetryMode fromString(String value) * Retry strategy applied to every AWS client built from this config. *

* Setting this at all is deliberate: left unset, the SDK picks its own default, and which one it picks depends on - * the {@code aws.newRetries2026} migration flag. Naming the mode here keeps retry behaviour stable across SDK + * the {@code aws.newRetries2026} migration flag. Naming the mode here keeps retry behavior stable across SDK * upgrades instead of changing under Druid when that flag's default flips. */ @JsonProperty From 2c57bb1a5cc125edcd4ccce5cb19f213b5fde3ef Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Wed, 5 Aug 2026 20:01:14 -0500 Subject: [PATCH 05/10] fix test --- .../java/org/apache/druid/data/input/s3/S3InputSourceTest.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java index 9840c4a68f97..bbc2df1fede0 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java @@ -78,6 +78,7 @@ import software.amazon.awssdk.core.ResponseInputStream; import software.amazon.awssdk.core.exception.SdkClientException; import software.amazon.awssdk.http.AbortableInputStream; +import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.S3ClientBuilder; import software.amazon.awssdk.services.s3.model.GetObjectRequest; @@ -519,6 +520,7 @@ public void testS3InputSourceUseEndPointClientProxy() EasyMock.expect(mockAwsClientConfig.getConnectionTimeoutMillis()).andStubReturn(10_000); EasyMock.expect(mockAwsClientConfig.getSocketTimeoutMillis()).andStubReturn(50_000); EasyMock.expect(mockAwsClientConfig.getMaxConnections()).andStubReturn(50); + EasyMock.expect(mockAwsClientConfig.getRetryStrategy()).andReturn(EasyMock.createMock(RetryStrategy.class)); EasyMock.expect(mockAwsProxyConfig.getHost()).andStubReturn(""); EasyMock.expect(mockAwsProxyConfig.getPort()).andStubReturn(-1); From 69b2b198d0c602f535d14bdd5c5b624b3f1f984f Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Thu, 6 Aug 2026 09:30:19 -0500 Subject: [PATCH 06/10] cleanup --- .../druid/common/aws/AWSClientConfig.java | 33 ++++------- .../druid/common/aws/AWSClientConfigTest.java | 2 +- docs/development/extensions-core/s3.md | 59 +------------------ 3 files changed, 14 insertions(+), 80 deletions(-) diff --git a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java index 3379a00f994f..c6ff78579d36 100644 --- a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java +++ b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java @@ -44,14 +44,6 @@ public class AWSClientConfig /** AWS SDK v2's own default. */ private static final int DEFAULT_MAX_CONNECTIONS_FLOOR = 50; - /** - * Selects the retry behavior AWS documents for {@code standard} and {@code adaptive} rather than the pre-2026 - * behavior. The no-argument factories default the {@code aws.newRetries2026} opt-in to false, which changes the - * backoff base delays and the retry-quota token costs; asking for it explicitly means the modes behave as - * documented, and keeps them behaving that way when AWS flips the opt-in default. - */ - private static final Boolean USE_DOCUMENTED_RETRY_BEHAVIOR = Boolean.TRUE; - /** * Retry strategy family. Declared as an enum so an unrecognised value is rejected while the config is bound at * startup, rather than when a client is first built. @@ -62,7 +54,8 @@ public enum RetryMode @Override RetryStrategy createStrategy() { - return AwsRetryStrategy.standardRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOR); + // Pass true to ensure we get the new standard AWS SDKv2 retry behavior and not legacy behavior. + return AwsRetryStrategy.standardRetryStrategy(true); } }, ADAPTIVE { @@ -72,7 +65,7 @@ RetryStrategy createStrategy() // Standard plus a client-side rate limiter, which unlike standard can delay or block the initial request, // not just retries. The limiter belongs to one client instance and covers every request that client makes, // so throttling on one key prefix also slows requests to prefixes that are not being throttled. - return AwsRetryStrategy.adaptiveRetryStrategy(USE_DOCUMENTED_RETRY_BEHAVIOR); + return AwsRetryStrategy.adaptiveRetryStrategy(true); } }, LEGACY { @@ -152,16 +145,12 @@ public static RetryMode fromString(String value) /** * Retry strategy applied to every AWS client built from this config. - *

- * Setting this at all is deliberate: left unset, the SDK picks its own default, and which one it picks depends on - * the {@code aws.newRetries2026} migration flag. Naming the mode here keeps retry behavior stable across SDK - * upgrades instead of changing under Druid when that flag's default flips. */ @JsonProperty private RetryMode retryMode = RetryMode.STANDARD; /** - * Total attempts per request, including the first. Maps directly to the SDK's {@code maxAttempts}, so 1 disables + * Total attempts per request, including the first. A value of 1 disables * retries. Null leaves the count that {@link #retryMode} defines for itself, which AWS tunes alongside that mode's * backoff and retry quota. *

@@ -172,7 +161,7 @@ public static RetryMode fromString(String value) @JsonProperty @Nullable @Min(1) - private Integer maxRetryAttempts = null; + private Integer maxAttempts = null; public String getProtocol() { @@ -246,9 +235,9 @@ public RetryMode getRetryMode() } @Nullable - public Integer getMaxRetryAttempts() + public Integer getMaxAttempts() { - return maxRetryAttempts; + return maxAttempts; } /** @@ -262,15 +251,15 @@ public RetryStrategy getRetryStrategy() } /** - * Overrides the attempt count only when one is configured, so an unset {@link #maxRetryAttempts} leaves whatever + * Overrides the attempt count only when one is configured, so an unset {@link #maxAttempts} leaves whatever * the chosen mode defines for itself. */ private RetryStrategy withMaxAttempts(RetryStrategy strategy) { - if (maxRetryAttempts == null) { + if (maxAttempts == null) { return strategy; } - return strategy.toBuilder().maxAttempts(maxRetryAttempts).build(); + return strategy.toBuilder().maxAttempts(maxAttempts).build(); } @Override @@ -285,7 +274,7 @@ public String toString() ", socketTimeout=" + socketTimeout + ", maxConnections=" + getMaxConnections() + ", retryMode='" + retryMode + '\'' + - ", maxRetryAttempts=" + maxRetryAttempts + + ", maxRetryAttempts=" + maxAttempts + '}'; } } diff --git a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java index 8058405f9020..6b484d5b2e6b 100644 --- a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java +++ b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java @@ -146,7 +146,7 @@ public void testUnsetAttemptCountLeavesTheCountTheModeDefines() { final AWSClientConfig config = new AWSClientConfig(); - Assertions.assertNull(config.getMaxRetryAttempts()); + Assertions.assertNull(config.getMaxAttempts()); Assertions.assertEquals( AWSClientConfig.RetryMode.STANDARD.createStrategy().maxAttempts(), config.getRetryStrategy().maxAttempts() diff --git a/docs/development/extensions-core/s3.md b/docs/development/extensions-core/s3.md index c6dd821182ad..d546cf4bf6a7 100644 --- a/docs/development/extensions-core/s3.md +++ b/docs/development/extensions-core/s3.md @@ -131,8 +131,8 @@ For example, to set the region to 'us-east-1' through system properties: |`druid.s3.disableChunkedEncoding`|Disables chunked encoding. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#disableChunkedEncoding--) for details.|false| |`druid.s3.enablePathStyleAccess`|Enables path style access. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#enablePathStyleAccess--) for details.|false| |`druid.s3.crossRegionAccessEnabled`|Enables cross-region access for S3 requests. When enabled, the S3 client automatically detects the correct region for a bucket on first access and caches it for subsequent requests.|false| -|`druid.s3.retryMode`|Retry strategy used by every AWS client Druid builds. One of `standard`, `adaptive` or `legacy`. See [Retry behavior](#retry-behavior).|`standard`| -|`druid.s3.maxRetryAttempts`|Total attempts per HTTP request, including the first, so `1` disables SDK retries. When unset, the count defined by `druid.s3.retryMode` applies. See [Retry behavior](#retry-behavior).|null (the retry mode's own default)| +|`druid.s3.retryMode`|Retry strategy used by every AWS client Druid builds. One of `standard`, `adaptive` or `legacy`. See [AWS document](https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html) for details about each mode.|`standard`| +|`druid.s3.maxAttempts`|Total attempts per HTTP request, including the first, so `1` disables SDK retries. When unset, the SDK default count defined by `druid.s3.retryMode` applies.|null (the retry mode's own default)| |`druid.s3.forceGlobalBucketAccessEnabled`|**Deprecated.** Use `druid.s3.crossRegionAccessEnabled` instead. Only used as a fallback if `crossRegionAccessEnabled` is not explicitly set.|null| |`druid.s3.endpoint.url`|Service endpoint either with or without the protocol.|None| |`druid.s3.endpoint.signingRegion`|Region to use for SigV4 signing of requests (e.g. us-west-1).|None| @@ -144,61 +144,6 @@ For example, to set the region to 'us-east-1' through system properties: |`druid.storage.sse.kms.keyId`|AWS KMS key ID. This is used only when `druid.storage.sse.type` is `kms` and can be empty to use the default key ID.|None| |`druid.storage.sse.custom.base64EncodedKey`|Base64-encoded key. Should be specified if `druid.storage.sse.type` is `custom`.|None| -## Retry behavior - -Druid names the AWS SDK retry strategy explicitly rather than accepting the SDK's default. The SDK's own default -depends on the `aws.newRetries2026` migration flag, so leaving it unset would let retry behavior change underneath -Druid when that flag's default flips. For `standard` and `adaptive`, Druid also opts in to the behavior AWS -documents for those modes, rather than quietly falling back to any legacy behavior the SDK may have. - -|Mode|Behavior| -|----|--------| -|`standard`|Error classification consistent with the other AWS SDK implementations, and the mode AWS recommends for all workloads.| -|`adaptive`|`standard` plus a client-side rate limiter that slows requests down when S3 reports throttling. Unlike `standard`, it can delay or block the **initial** request, not only retries. The limiter covers every request made by one client instance, so throttling on one key prefix also slows requests to prefixes that are not being throttled.| -|`legacy`|The SDK's legacy behavior, retained so a deployment can revert without a rollback. AWS recommends moving off it.| - -### Choosing a mode - -AWS recommends `standard` as the default and `adaptive` only for workloads that are *single-resource, -throttling-heavy, and latency-tolerant*. S3 applies its request-rate limits per key prefix, so "single-resource" -means a process that concentrates its requests on one prefix. - -That maps onto Druid roughly as follows. Because a peon runs one task, you can select a mode per task type — through -a Kubernetes pod template, or `druid.indexer.fork.property.druid.s3.retryMode` in a task's context when using the -MiddleManager task runner. - -|Process|Suggested mode|Reasoning| -|-------|--------------|---------| -|MSQ and compaction peons|`adaptive`|Segment output goes to one new version prefix and shuffle output to one prefix per query, which is exactly the concentration that provokes throttling. Batch work tolerates the added latency.| -|Batch ingestion peons|`standard`, or `adaptive` if throttled|Same shape as above at lower request rates.| -|Historicals|`standard`|One client loads segments for every datasource the process serves, so a rate limiter tripped by one prefix would slow loads for unrelated ones. Segment loads can also sit on the query path.| -|Brokers, Coordinator, Overlord|`standard`|Low request volume, and delaying an initial request costs query latency for no benefit.| - -### Retry quota - -Every mode carries a retry quota: a token bucket, held per client instance and never shared across processes, that -stops retries once it is exhausted so the client fails fast instead of adding load a struggling service cannot -absorb. It only engages under sustained failure — roughly a 32% failure rate for throttling errors — and is inert -otherwise. Choosing between the modes does not change whether it is present. - -S3 reports throttling as `SlowDown` and `503`, which the SDK classifies as throttling rather than transient errors. -Those get a longer base backoff than transient failures, and they are what `adaptive`'s rate limiter reacts to. - -### Retries are layered - -`druid.s3.maxRetryAttempts` applies per HTTP request. Druid retries again on top of it, and the two multiply: - -|Layer|Scope of one attempt|Attempts|Backoff cap| -|-----|--------------------|--------|-----------| -|`druid.s3.maxRetryAttempts`|A single HTTP request, such as one `UploadPart`|Set by the retry mode|`~20s`| -|`S3Utils.retryS3Operation`|A whole logical operation, such as re-uploading an entire segment|10|`60s`| -|`druid.msq.intermediate.storage.maxRetry`|A durable storage operation|10|`60s`| - -The layers are not equivalent, and the difference matters more than the totals: an attempt at the first layer -re-sends one part, whereas a retry at the layers below it repeats the whole transfer. A workload that keeps losing -uploads to throttling is usually better served by raising `druid.s3.maxRetryAttempts`, which absorbs failures before -they reach a layer that re-uploads everything, than by raising the Druid-level counts. - ## Server-side encryption You can enable [server-side encryption](https://docs.aws.amazon.com/AmazonS3/latest/dev/serv-side-encryption) by setting From a1b4286abcb8d7af2e9064ae8a72530444fa361e Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Thu, 6 Aug 2026 15:55:38 -0500 Subject: [PATCH 07/10] make retry strategy instance per client + ensure clients from things like s3 input source get strategy when needed --- .../druid/data/input/s3/S3InputSource.java | 2 +- .../s3/ServerSideEncryptingAmazonS3.java | 48 +++++++++++++---- .../data/input/s3/S3InputSourceTest.java | 52 ++++++++++++++++++- .../s3/ServerSideEncryptingAmazonS3Test.java | 34 ++++++++++++ 4 files changed, 124 insertions(+), 12 deletions(-) diff --git a/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java b/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java index 85cd832d98f9..3d51ea73d876 100644 --- a/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java +++ b/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java @@ -127,7 +127,7 @@ public S3InputSource( s3ClientBuilder.getS3StorageConfig(), awsProxyConfig, awsEndpointConfig, - awsClientConfig, + awsClientConfig != null ? awsClientConfig : s3ClientBuilder.getAwsClientConfig(), s3InputSourceConfig, null ).build(); 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 8c356111680c..b56f07c7c85e 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 @@ -375,16 +375,12 @@ public static ServerSideEncryptingAmazonS3.Builder builder( S3Configuration s3Config = S3Configuration.builder() .chunkedEncodingEnabled(!awsClientConfig.isDisableChunkedEncoding()) .build(); - final ClientOverrideConfiguration retryOverrides = - ClientOverrideConfiguration.builder() - .retryStrategy(awsClientConfig.getRetryStrategy()) - .build(); clientBuilder.serviceConfiguration(s3Config) .forcePathStyle(awsClientConfig.isEnablePathStyleAccess()) .crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled()) - .overrideConfiguration(retryOverrides); + .overrideConfiguration(retryOverride(awsClientConfig)); final S3TransferConfig transferConfig = s3StorageConfig.getS3TransferConfig(); - asyncClientBuilder.overrideConfiguration(retryOverrides) + asyncClientBuilder.overrideConfiguration(retryOverride(awsClientConfig)) .forcePathStyle(awsClientConfig.isEnablePathStyleAccess()) .crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled()) .httpClientBuilder(AsyncHttpClientType.fromString(transferConfig.getAsyncHttpClientType()).buildBuilder(awsClientConfig)) @@ -430,7 +426,19 @@ public static ServerSideEncryptingAmazonS3.Builder builder( return ServerSideEncryptingAmazonS3.builder() .setS3ClientSupplier(clientBuilder::build) .setS3AsyncClientSupplier(asyncClientBuilder::build) - .setS3StorageConfig(s3StorageConfig); + .setS3StorageConfig(s3StorageConfig) + .setAwsClientConfig(awsClientConfig); + } + + /** + * Builds the retry override for a single client. Every client needs its own {@code RetryStrategy} instance: the + * standard strategy holds its circuit-breaker token bucket on the instance, and the adaptive strategy additionally + * holds its client-side rate limiter there. Sharing one instance would let throttled TransferManager uploads drain + * the quota of, or throttle, synchronous reads and listings. + */ + private static ClientOverrideConfiguration retryOverride(AWSClientConfig awsClientConfig) + { + return ClientOverrideConfiguration.builder().retryStrategy(awsClientConfig.getRetryStrategy()).build(); } @Nonnull @@ -468,10 +476,10 @@ public static AwsCredentialsProvider createAssumeRoleCredentialsProvider( if (awsEndpointConfig != null && awsEndpointConfig.getSigningRegion() != null) { stsBuilder.region(Region.of(awsEndpointConfig.getSigningRegion())); } + // A null config leaves the SDK defaults in place, so callers should pass the process configuration rather than + // null when a spec supplies no override of its own. if (awsClientConfig != null) { - stsBuilder.overrideConfiguration( - ClientOverrideConfiguration.builder().retryStrategy(awsClientConfig.getRetryStrategy()).build() - ); + stsBuilder.overrideConfiguration(retryOverride(awsClientConfig)); } AssumeRoleRequest.Builder assumeRoleRequestBuilder = @@ -494,6 +502,8 @@ public static class Builder @Nullable private Supplier s3AsyncClientSupplier; private S3StorageConfig s3StorageConfig = new S3StorageConfig(new NoopServerSideEncryption(), null); + @Nullable + private AWSClientConfig awsClientConfig; public Builder setS3ClientSupplier(Supplier s3ClientSupplier) { @@ -518,6 +528,24 @@ public S3StorageConfig getS3StorageConfig() return this.s3StorageConfig; } + public Builder setAwsClientConfig(@Nullable AWSClientConfig awsClientConfig) + { + this.awsClientConfig = awsClientConfig; + return this; + } + + /** + * The client configuration the clients of this builder are configured with. Exposed, like + * {@link #getS3StorageConfig()}, so a caller that rebuilds a client for a per-spec override can fall back to it + * instead of dropping process-level settings. Null when the builder was not + * created from an {@link AWSClientConfig}, in which case the clients keep the SDK defaults. + */ + @Nullable + public AWSClientConfig getAwsClientConfig() + { + return this.awsClientConfig; + } + /** * Builds a new {@link ServerSideEncryptingAmazonS3} instance. * diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java index bbc2df1fede0..fbac2a76071c 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java @@ -100,6 +100,7 @@ import java.util.Collections; import java.util.EnumSet; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -348,6 +349,7 @@ public void testSerdeWithCloudConfigPropertiesWithKeyAndSecret() throws Exceptio { EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); + EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final S3InputSource withPrefixes = new S3InputSource( SERVICE, @@ -375,6 +377,7 @@ public void testSerdeWithCloudConfigPropertiesWithSessionToken() throws Exceptio { EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); + EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final S3InputSource withSessionToken = new S3InputSource( SERVICE, @@ -407,6 +410,7 @@ public void testSchemelessEndpointConfigUrlWithNullClientConfigResolvesSupplier( EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()) .andStubReturn(S3_STORAGE_CONFIG); + EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final AWSEndpointConfig schemelessEndpoint = MAPPER.readValue( @@ -434,6 +438,49 @@ public void testSchemelessEndpointConfigUrlWithNullClientConfigResolvesSupplier( EasyMock.verify(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); } + @Test + public void testClientConfigOmittedFromSpecFallsBackToProcessConfig() + { + // A spec that overrides credentials but not the client config must still get the process-wide client settings + final AtomicInteger strategiesRequested = new AtomicInteger(); + final AWSClientConfig processClientConfig = new AWSClientConfig() + { + @Override + public RetryStrategy getRetryStrategy() + { + strategiesRequested.incrementAndGet(); + return super.getRetryStrategy(); + } + }; + + EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); + EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); + EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()) + .andStubReturn(processClientConfig); + EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); + + final S3InputSource inputSource = new S3InputSource( + SERVICE, + SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER, + INPUT_DATA_CONFIG, + null, + null, + EXPECTED_LOCATION, + null, + CLOUD_CONFIG_PROPERTIES, + null, + ENDPOINT_CONFIG, + null + ); + + // Forces s3ClientSupplier evaluation, which is where the clients are configured. + inputSource.createEntity(new CloudObjectLocation("bucket", "path")); + + // One per client, sync and async. + Assertions.assertEquals(2, strategiesRequested.get()); + EasyMock.verify(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); + } + @Test public void testGetSetSessionToken() { @@ -520,7 +567,10 @@ public void testS3InputSourceUseEndPointClientProxy() EasyMock.expect(mockAwsClientConfig.getConnectionTimeoutMillis()).andStubReturn(10_000); EasyMock.expect(mockAwsClientConfig.getSocketTimeoutMillis()).andStubReturn(50_000); EasyMock.expect(mockAwsClientConfig.getMaxConnections()).andStubReturn(50); - EasyMock.expect(mockAwsClientConfig.getRetryStrategy()).andReturn(EasyMock.createMock(RetryStrategy.class)); + // Once for the sync client and once for the async one, since the two must not share a strategy instance. + EasyMock.expect(mockAwsClientConfig.getRetryStrategy()) + .andReturn(EasyMock.createMock(RetryStrategy.class)) + .times(2); EasyMock.expect(mockAwsProxyConfig.getHost()).andStubReturn(""); EasyMock.expect(mockAwsProxyConfig.getPort()).andStubReturn(-1); diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3Test.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3Test.java index acc93a159dff..6f16924c5244 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3Test.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/ServerSideEncryptingAmazonS3Test.java @@ -19,13 +19,16 @@ package org.apache.druid.storage.s3; +import org.apache.druid.common.aws.AWSClientConfig; import org.easymock.Capture; import org.easymock.EasyMock; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.model.Grant; @@ -40,6 +43,8 @@ import java.io.File; import java.io.IOException; import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CompletableFuture; @@ -255,6 +260,35 @@ public void testBuilder() Assertions.assertEquals(builtClient, s3.getS3Client()); } + @Test + public void testEachClientGetsItsOwnRetryStrategy() + { + final List issued = new ArrayList<>(); + final AWSClientConfig clientConfig = new AWSClientConfig() + { + @Override + public RetryStrategy getRetryStrategy() + { + final RetryStrategy strategy = super.getRetryStrategy(); + issued.add(strategy); + return strategy; + } + }; + + ServerSideEncryptingAmazonS3.builder( + AnonymousCredentialsProvider.create(), + new S3StorageConfig(new NoopServerSideEncryption(), new S3TransferConfig()), + null, + null, + clientConfig, + null, + null + ); + + Assertions.assertEquals(2, issued.size(), "one strategy per client, sync and async"); + Assertions.assertNotSame(issued.get(0), issued.get(1)); + } + @Test public void testBuilderWithAsyncClient() throws NoSuchFieldException, IllegalAccessException { From b0b3472169e6b4b18d33e1a68a0c568b1222390d Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Thu, 6 Aug 2026 16:16:05 -0500 Subject: [PATCH 08/10] actually don't do the 2nd comment suggestion --- docs/development/extensions-core/s3.md | 2 +- .../druid/data/input/s3/S3InputSource.java | 2 +- .../s3/ServerSideEncryptingAmazonS3.java | 25 +--------- .../data/input/s3/S3InputSourceTest.java | 47 ------------------- 4 files changed, 3 insertions(+), 73 deletions(-) diff --git a/docs/development/extensions-core/s3.md b/docs/development/extensions-core/s3.md index a6b1264a27a4..b60e80585e51 100644 --- a/docs/development/extensions-core/s3.md +++ b/docs/development/extensions-core/s3.md @@ -131,7 +131,7 @@ For example, to set the region to 'us-east-1' through system properties: |`druid.s3.disableChunkedEncoding`|Disables chunked encoding. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#disableChunkedEncoding--) for details.|false| |`druid.s3.enablePathStyleAccess`|Enables path style access. See [AWS document](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#enablePathStyleAccess--) for details.|false| |`druid.s3.crossRegionAccessEnabled`|Enables cross-region access for S3 requests. When enabled, the S3 client automatically detects the correct region for a bucket on first access and caches it for subsequent requests.|false| -|`druid.s3.retryMode`|Retry strategy used by every AWS client Druid builds. One of `standard`, `adaptive` or `legacy`. See [AWS document](https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html) for details about each mode.|`standard`| +|`druid.s3.retryMode`|Retry strategy for AWS clients built from this configuration. One of `standard`, `adaptive` or `legacy`. See [AWS document](https://docs.aws.amazon.com/sdkref/latest/guide/feature-retry-behavior.html) for details about each mode.|`standard`| |`druid.s3.maxAttempts`|Total attempts per HTTP request, including the first, so `1` disables SDK retries. When unset, the SDK default count defined by `druid.s3.retryMode` applies.|null (the retry mode's own default)| |`druid.s3.forceGlobalBucketAccessEnabled`|**Deprecated.** Use `druid.s3.crossRegionAccessEnabled` instead. Only used as a fallback if `crossRegionAccessEnabled` is not explicitly set.|null| |`druid.s3.endpoint.url`|Service endpoint either with or without the protocol.|None| diff --git a/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java b/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java index 3d51ea73d876..85cd832d98f9 100644 --- a/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java +++ b/extensions-core/s3-extensions/src/main/java/org/apache/druid/data/input/s3/S3InputSource.java @@ -127,7 +127,7 @@ public S3InputSource( s3ClientBuilder.getS3StorageConfig(), awsProxyConfig, awsEndpointConfig, - awsClientConfig != null ? awsClientConfig : s3ClientBuilder.getAwsClientConfig(), + awsClientConfig, s3InputSourceConfig, null ).build(); 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 b56f07c7c85e..b1e6cb2f0439 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 @@ -426,8 +426,7 @@ public static ServerSideEncryptingAmazonS3.Builder builder( return ServerSideEncryptingAmazonS3.builder() .setS3ClientSupplier(clientBuilder::build) .setS3AsyncClientSupplier(asyncClientBuilder::build) - .setS3StorageConfig(s3StorageConfig) - .setAwsClientConfig(awsClientConfig); + .setS3StorageConfig(s3StorageConfig); } /** @@ -476,8 +475,6 @@ public static AwsCredentialsProvider createAssumeRoleCredentialsProvider( if (awsEndpointConfig != null && awsEndpointConfig.getSigningRegion() != null) { stsBuilder.region(Region.of(awsEndpointConfig.getSigningRegion())); } - // A null config leaves the SDK defaults in place, so callers should pass the process configuration rather than - // null when a spec supplies no override of its own. if (awsClientConfig != null) { stsBuilder.overrideConfiguration(retryOverride(awsClientConfig)); } @@ -502,8 +499,6 @@ public static class Builder @Nullable private Supplier s3AsyncClientSupplier; private S3StorageConfig s3StorageConfig = new S3StorageConfig(new NoopServerSideEncryption(), null); - @Nullable - private AWSClientConfig awsClientConfig; public Builder setS3ClientSupplier(Supplier s3ClientSupplier) { @@ -528,24 +523,6 @@ public S3StorageConfig getS3StorageConfig() return this.s3StorageConfig; } - public Builder setAwsClientConfig(@Nullable AWSClientConfig awsClientConfig) - { - this.awsClientConfig = awsClientConfig; - return this; - } - - /** - * The client configuration the clients of this builder are configured with. Exposed, like - * {@link #getS3StorageConfig()}, so a caller that rebuilds a client for a per-spec override can fall back to it - * instead of dropping process-level settings. Null when the builder was not - * created from an {@link AWSClientConfig}, in which case the clients keep the SDK defaults. - */ - @Nullable - public AWSClientConfig getAwsClientConfig() - { - return this.awsClientConfig; - } - /** * Builds a new {@link ServerSideEncryptingAmazonS3} instance. * diff --git a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java index fbac2a76071c..ba3912c49a32 100644 --- a/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java +++ b/extensions-core/s3-extensions/src/test/java/org/apache/druid/data/input/s3/S3InputSourceTest.java @@ -100,7 +100,6 @@ import java.util.Collections; import java.util.EnumSet; import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -349,7 +348,6 @@ public void testSerdeWithCloudConfigPropertiesWithKeyAndSecret() throws Exceptio { EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); - EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final S3InputSource withPrefixes = new S3InputSource( SERVICE, @@ -377,7 +375,6 @@ public void testSerdeWithCloudConfigPropertiesWithSessionToken() throws Exceptio { EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); - EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final S3InputSource withSessionToken = new S3InputSource( SERVICE, @@ -410,7 +407,6 @@ public void testSchemelessEndpointConfigUrlWithNullClientConfigResolvesSupplier( EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()) .andStubReturn(S3_STORAGE_CONFIG); - EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()).andStubReturn(null); EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); final AWSEndpointConfig schemelessEndpoint = MAPPER.readValue( @@ -438,49 +434,6 @@ public void testSchemelessEndpointConfigUrlWithNullClientConfigResolvesSupplier( EasyMock.verify(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); } - @Test - public void testClientConfigOmittedFromSpecFallsBackToProcessConfig() - { - // A spec that overrides credentials but not the client config must still get the process-wide client settings - final AtomicInteger strategiesRequested = new AtomicInteger(); - final AWSClientConfig processClientConfig = new AWSClientConfig() - { - @Override - public RetryStrategy getRetryStrategy() - { - strategiesRequested.incrementAndGet(); - return super.getRetryStrategy(); - } - }; - - EasyMock.reset(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); - EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getS3StorageConfig()).andStubReturn(S3_STORAGE_CONFIG); - EasyMock.expect(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER.getAwsClientConfig()) - .andStubReturn(processClientConfig); - EasyMock.replay(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); - - final S3InputSource inputSource = new S3InputSource( - SERVICE, - SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER, - INPUT_DATA_CONFIG, - null, - null, - EXPECTED_LOCATION, - null, - CLOUD_CONFIG_PROPERTIES, - null, - ENDPOINT_CONFIG, - null - ); - - // Forces s3ClientSupplier evaluation, which is where the clients are configured. - inputSource.createEntity(new CloudObjectLocation("bucket", "path")); - - // One per client, sync and async. - Assertions.assertEquals(2, strategiesRequested.get()); - EasyMock.verify(SERVER_SIDE_ENCRYPTING_AMAZON_S3_BUILDER); - } - @Test public void testGetSetSessionToken() { From fb241a899a43731059bf6da08ee563c600c1ebee Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Thu, 6 Aug 2026 16:22:59 -0500 Subject: [PATCH 09/10] get explicit on saying the retry strategy object shouldnt be shared --- .../main/java/org/apache/druid/common/aws/AWSClientConfig.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java index c6ff78579d36..67f31cfbe981 100644 --- a/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java +++ b/cloud/aws-common/src/main/java/org/apache/druid/common/aws/AWSClientConfig.java @@ -244,6 +244,9 @@ public Integer getMaxAttempts() * Builds the strategy to hand to {@code ClientOverrideConfiguration.retryStrategy}. Kept as a plain function of the * config because a built AWS client does not expose the strategy it was given, so this is the only place the * mapping can be tested. + *

+ * Returns a new instance per call; clients must not share one, since the strategies hold their circuit-breaker + * quota (and, for adaptive, their rate limiter) on the instance. */ public RetryStrategy getRetryStrategy() { From 9840a144872b9503e609e01fb9bfc7a378ae0171 Mon Sep 17 00:00:00 2001 From: Lucas Capistrant Date: Fri, 7 Aug 2026 08:30:35 -0500 Subject: [PATCH 10/10] fix test --- .../java/org/apache/druid/common/aws/AWSClientConfigTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java index 6b484d5b2e6b..a86a2900af86 100644 --- a/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java +++ b/cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientConfigTest.java @@ -156,7 +156,7 @@ public void testUnsetAttemptCountLeavesTheCountTheModeDefines() @Test public void testConfiguredAttemptCountIsApplied() { - Assertions.assertEquals(8, bind(Map.of("maxRetryAttempts", 8)).getRetryStrategy().maxAttempts()); + Assertions.assertEquals(8, bind(Map.of("maxAttempts", 8)).getRetryStrategy().maxAttempts()); } @ParameterizedTest(name = "{0}")