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 468d2dd2ce4 feat: Add operator access to configure AWS retry policy.
Default to STANDARD, recommended by AWS (#19892)
468d2dd2ce4 is described below
commit 468d2dd2ce4e689a15b2738cb28fbd04e3117b5d
Author: Lucas Capistrant <[email protected]>
AuthorDate: Fri Aug 7 10:08:24 2026 -0500
feat: Add operator access to configure AWS retry policy. Default to
STANDARD, recommended by AWS (#19892)
---
cloud/aws-common/pom.xml | 16 ++
.../apache/druid/common/aws/AWSClientConfig.java | 121 ++++++++++++++
.../druid/common/aws/AWSClientConfigTest.java | 177 +++++++++++++++------
docs/development/extensions-core/s3.md | 2 +
extensions-core/s3-extensions/pom.xml | 5 +
.../storage/s3/ServerSideEncryptingAmazonS3.java | 27 +++-
.../druid/data/input/s3/S3InputSourceTest.java | 5 +
.../s3/ServerSideEncryptingAmazonS3Test.java | 34 ++++
8 files changed, 338 insertions(+), 49 deletions(-)
diff --git a/cloud/aws-common/pom.xml b/cloud/aws-common/pom.xml
index d1197a934d5..ab664679924 100644
--- a/cloud/aws-common/pom.xml
+++ b/cloud/aws-common/pom.xml
@@ -82,6 +82,17 @@
<artifactId>sdk-core</artifactId>
<version>${aws.sdk.v2.version}</version>
</dependency>
+ <!-- Retry strategies configured by AWSClientConfig -->
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>retries-spi</artifactId>
+ <version>${aws.sdk.v2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>retries</artifactId>
+ <version>${aws.sdk.v2.version}</version>
+ </dependency>
<!-- WebIdentityTokenProvider requires runtime dependency on sts -->
<dependency>
<groupId>software.amazon.awssdk</groupId>
@@ -115,5 +126,10 @@
<artifactId>junit-jupiter-api</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-params</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
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 9fdf9ba592f..67f31cfbe98 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,61 @@ public class AWSClientConfig
/** AWS SDK v2's own default. */
private static final int DEFAULT_MAX_CONNECTIONS_FLOOR = 50;
+ /**
+ * 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()
+ {
+ // Pass true to ensure we get the new standard AWS SDKv2 retry
behavior and not legacy behavior.
+ return AwsRetryStrategy.standardRetryStrategy(true);
+ }
+ },
+ 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(true);
+ }
+ },
+ LEGACY {
+ @Override
+ RetryStrategy createStrategy()
+ {
+ // 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();
+ }
+ };
+
+ 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 +143,26 @@ public class AWSClientConfig
@Nullable
private Integer maxConnections = null;
+ /**
+ * Retry strategy applied to every AWS client built from this config.
+ */
+ @JsonProperty
+ private RetryMode retryMode = RetryMode.STANDARD;
+
+ /**
+ * 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.
+ * <p>
+ * 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 maxAttempts = null;
+
public String getProtocol()
{
return protocol;
@@ -146,6 +229,42 @@ public class AWSClientConfig
return Math.max(DEFAULT_MAX_CONNECTIONS_FLOOR, 4 *
runtimeInfo.getAvailableProcessors());
}
+ public RetryMode getRetryMode()
+ {
+ return retryMode;
+ }
+
+ @Nullable
+ public Integer getMaxAttempts()
+ {
+ return maxAttempts;
+ }
+
+ /**
+ * 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.
+ * <p>
+ * 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()
+ {
+ return withMaxAttempts(retryMode.createStrategy());
+ }
+
+ /**
+ * 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 (maxAttempts == null) {
+ return strategy;
+ }
+ return strategy.toBuilder().maxAttempts(maxAttempts).build();
+ }
+
@Override
public String toString()
{
@@ -157,6 +276,8 @@ public class AWSClientConfig
", connectionTimeout=" + connectionTimeout +
", socketTimeout=" + socketTimeout +
", maxConnections=" + getMaxConnections() +
+ ", retryMode='" + retryMode + '\'' +
+ ", 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 99b927efb2b..a86a2900af8 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
@@ -21,15 +21,42 @@ package org.apache.druid.common.aws;
import com.fasterxml.jackson.databind.InjectableValues;
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 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<String, Object> properties)
+ {
+ return MAPPER.convertValue(properties, AWSClientConfig.class);
+ }
+
+ private static AWSClientConfig bind(Map<String, Object> properties,
RuntimeInfo runtimeInfo)
+ {
+ return mapperWithRuntimeInfo(runtimeInfo).convertValue(properties,
AWSClientConfig.class);
+ }
private static ObjectMapper mapperWithRuntimeInfo(RuntimeInfo runtimeInfo)
{
@@ -39,84 +66,144 @@ public class AWSClientConfigTest
}
@Test
- public void testDefaultCrossRegionAccessEnabled() throws Exception
+ public void testDefaultRetryModeIsStandard()
{
- AWSClientConfig config = MAPPER.readValue("{}", AWSClientConfig.class);
- Assertions.assertNull(config.isForceGlobalBucketAccessEnabled());
- Assertions.assertFalse(config.isCrossRegionAccessEnabled());
+ final AWSClientConfig config = new AWSClientConfig();
+
+ Assertions.assertEquals(AWSClientConfig.RetryMode.STANDARD,
config.getRetryMode());
+ Assertions.assertInstanceOf(StandardRetryStrategy.class,
config.getRetryStrategy());
}
- @Test
- public void testCrossRegionAccessEnabledExplicitlySet() throws Exception
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("retryModeStrategies")
+ public void testEachRetryModeBuildsItsStrategy(
+ AWSClientConfig.RetryMode mode,
+ Class<? extends RetryStrategy> expected
+ )
{
- AWSClientConfig config = MAPPER.readValue("{\"crossRegionAccessEnabled\":
true}", AWSClientConfig.class);
- Assertions.assertNull(config.isForceGlobalBucketAccessEnabled());
- Assertions.assertTrue(config.isCrossRegionAccessEnabled());
+ Assertions.assertInstanceOf(expected, mode.createStrategy());
}
+ /**
+ * Guards {@link #retryModeStrategies} against a mode being added without a
strategy expectation.
+ */
@Test
- public void testNewConfigTakesPrecedenceOverDeprecatedWhenBothSet() throws
Exception
+ public void testEveryRetryModeHasAStrategyExpectation()
{
- AWSClientConfig config = MAPPER.readValue(
- "{\"forceGlobalBucketAccessEnabled\": true,
\"crossRegionAccessEnabled\": false}",
- AWSClientConfig.class
+ Assertions.assertEquals(AWSClientConfig.RetryMode.values().length,
retryModeStrategies().count());
+ }
+
+ private static Stream<Arguments> retryModeStrategies()
+ {
+ 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.assertFalse(config.isCrossRegionAccessEnabled());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"adaptive", "ADAPTIVE", "Adaptive"})
+ public void testRetryModeParsingIsCaseInsensitive(String value)
+ {
+ Assertions.assertEquals(AWSClientConfig.RetryMode.ADAPTIVE,
AWSClientConfig.RetryMode.fromString(value));
}
@Test
- public void testNewConfigTrueWinsOverDeprecatedFalse() throws Exception
+ public void testRetryModeBindsFromItsProperty()
{
- AWSClientConfig config = MAPPER.readValue(
- "{\"forceGlobalBucketAccessEnabled\": false,
\"crossRegionAccessEnabled\": true}",
- AWSClientConfig.class
+ Assertions.assertEquals(
+ AWSClientConfig.RetryMode.ADAPTIVE,
+ bind(Map.of("retryMode", "adaptive")).getRetryMode()
);
- Assertions.assertTrue(config.isCrossRegionAccessEnabled());
}
@Test
- public void testDeprecatedForceGlobalBucketAccessAloneTrue() throws Exception
+ public void testRetryModeSerializesToItsPropertyValue()
+ {
+ Assertions.assertEquals("adaptive",
MAPPER.convertValue(AWSClientConfig.RetryMode.ADAPTIVE, String.class));
+ }
+
+ /**
+ * 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()
{
- AWSClientConfig config = MAPPER.readValue(
- "{\"forceGlobalBucketAccessEnabled\": true}",
- AWSClientConfig.class
+ final IllegalArgumentException e = Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> bind(Map.of("retryMode", "aggressive"))
);
- Assertions.assertTrue(config.isCrossRegionAccessEnabled());
+
+ final Throwable rootCause = Throwables.getRootCause(e);
+ Assertions.assertInstanceOf(IAE.class, rootCause);
+ Assertions.assertTrue(rootCause.getMessage().contains("aggressive"));
}
@Test
- public void testDeprecatedNotSetFallsThroughToCrossRegion() throws Exception
+ public void testUnsetAttemptCountLeavesTheCountTheModeDefines()
{
- AWSClientConfig config = MAPPER.readValue(
- "{\"crossRegionAccessEnabled\": true}",
- AWSClientConfig.class
+ final AWSClientConfig config = new AWSClientConfig();
+
+ Assertions.assertNull(config.getMaxAttempts());
+ Assertions.assertEquals(
+ AWSClientConfig.RetryMode.STANDARD.createStrategy().maxAttempts(),
+ config.getRetryStrategy().maxAttempts()
);
- Assertions.assertNull(config.isForceGlobalBucketAccessEnabled());
- Assertions.assertTrue(config.isCrossRegionAccessEnabled());
}
@Test
- public void testDefaultMaxConnectionsKeepsAwsSdkFloorOnSmallHost() throws
Exception
+ public void testConfiguredAttemptCountIsApplied()
{
- AWSClientConfig config = mapperWithRuntimeInfo(new
FixedProcessorsRuntimeInfo(8))
- .readValue("{}", AWSClientConfig.class);
- Assertions.assertEquals(50, config.getMaxConnections());
+ Assertions.assertEquals(8, bind(Map.of("maxAttempts",
8)).getRetryStrategy().maxAttempts());
}
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("crossRegionAccessBindings")
+ public void testCrossRegionAccessResolution(Map<String, Object> properties,
boolean expected)
+ {
+ Assertions.assertEquals(expected,
bind(properties).isCrossRegionAccessEnabled());
+ }
+
+ private static Stream<Arguments> crossRegionAccessBindings()
+ {
+ 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)
+ );
+ }
+
+ /**
+ * 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 testDefaultMaxConnectionsScalesWithCoresOnLargeHost() throws
Exception
+ @SuppressWarnings("deprecation")
+ public void testDeprecatedPropertyStaysUnsetWhenOnlyItsReplacementIsBound()
{
- AWSClientConfig config = mapperWithRuntimeInfo(new
FixedProcessorsRuntimeInfo(32))
- .readValue("{}", AWSClientConfig.class);
- Assertions.assertEquals(128, config.getMaxConnections());
+ Assertions.assertNull(bind(Map.of()).isForceGlobalBucketAccessEnabled());
+ Assertions.assertNull(bind(Map.of("crossRegionAccessEnabled",
true)).isForceGlobalBucketAccessEnabled());
+ }
+
+ @ParameterizedTest(name = "{0} processors -> {1} connections")
+ @CsvSource({"8, 50", "32, 128"})
+ public void testDefaultMaxConnectionsTakesTheSdkFloorOrFourPerCore(int
processors, int expected)
+ {
+ 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
diff --git a/docs/development/extensions-core/s3.md
b/docs/development/extensions-core/s3.md
index c59f12670fd..b60e80585e5 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 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|
|`druid.s3.endpoint.signingRegion`|Region to use for SigV4 signing of requests
(e.g. us-west-1).|None|
diff --git a/extensions-core/s3-extensions/pom.xml
b/extensions-core/s3-extensions/pom.xml
index 40b3b549fd5..9b6581e8434 100644
--- a/extensions-core/s3-extensions/pom.xml
+++ b/extensions-core/s3-extensions/pom.xml
@@ -163,6 +163,11 @@
<artifactId>utils</artifactId>
<version>${aws.sdk.v2.version}</version>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>retries-spi</artifactId>
+ <version>${aws.sdk.v2.version}</version>
+ </dependency>
<dependency>
<groupId>it.unimi.dsi</groupId>
<artifactId>fastutil-core</artifactId>
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 dd86e9e610d..b1e6cb2f043 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.AwsCredentialsProvider;
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;
@@ -376,9 +377,11 @@ public class ServerSideEncryptingAmazonS3
.build();
clientBuilder.serviceConfiguration(s3Config)
.forcePathStyle(awsClientConfig.isEnablePathStyleAccess())
-
.crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled());
+
.crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled())
+ .overrideConfiguration(retryOverride(awsClientConfig));
final S3TransferConfig transferConfig =
s3StorageConfig.getS3TransferConfig();
-
asyncClientBuilder.forcePathStyle(awsClientConfig.isEnablePathStyleAccess())
+ asyncClientBuilder.overrideConfiguration(retryOverride(awsClientConfig))
+
.forcePathStyle(awsClientConfig.isEnablePathStyleAccess())
.crossRegionAccessEnabled(awsClientConfig.isCrossRegionAccessEnabled())
.httpClientBuilder(AsyncHttpClientType.fromString(transferConfig.getAsyncHttpClientType()).buildBuilder(awsClientConfig))
.multipartEnabled(true)
@@ -411,7 +414,8 @@ public class ServerSideEncryptingAmazonS3
assumeRoleArn,
assumeRoleExternalId,
awsEndpointConfig,
- credentialsProvider
+ credentialsProvider,
+ awsClientConfig
);
}
@@ -425,6 +429,17 @@ public class ServerSideEncryptingAmazonS3
.setS3StorageConfig(s3StorageConfig);
}
+ /**
+ * 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
private static StaticCredentialsProvider
createStaticCredentialsProvider(S3InputSourceConfig s3InputSourceConfig)
{
@@ -449,7 +464,8 @@ public class ServerSideEncryptingAmazonS3
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());
@@ -459,6 +475,9 @@ public class ServerSideEncryptingAmazonS3
if (awsEndpointConfig != null && awsEndpointConfig.getSigningRegion() !=
null) {
stsBuilder.region(Region.of(awsEndpointConfig.getSigningRegion()));
}
+ if (awsClientConfig != null) {
+ stsBuilder.overrideConfiguration(retryOverride(awsClientConfig));
+ }
AssumeRoleRequest.Builder assumeRoleRequestBuilder =
AssumeRoleRequest.builder().roleArn(assumeRoleArn).roleSessionName(roleSessionName).durationSeconds(3600);
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 9840c4a68f9..ba3912c49a3 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.auth.credentials.AwsCredentialsProvider;
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,10 @@ public class S3InputSourceTest extends
InitializedNullHandlingTest
EasyMock.expect(mockAwsClientConfig.getConnectionTimeoutMillis()).andStubReturn(10_000);
EasyMock.expect(mockAwsClientConfig.getSocketTimeoutMillis()).andStubReturn(50_000);
EasyMock.expect(mockAwsClientConfig.getMaxConnections()).andStubReturn(50);
+ // 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 acc93a159df..6f16924c524 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
software.amazon.awssdk.transfer.s3.model.UploadFileRequest;
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 class ServerSideEncryptingAmazonS3Test
Assertions.assertEquals(builtClient, s3.getS3Client());
}
+ @Test
+ public void testEachClientGetsItsOwnRetryStrategy()
+ {
+ final List<RetryStrategy> 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
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
