diff --git a/cloud/aws-common/pom.xml b/cloud/aws-common/pom.xml
index d1197a934d58..ab6646799244 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
@@ -115,5 +126,10 @@
junit-jupiter-api
test
+
+ org.junit.jupiter
+ junit-jupiter-params
+ test
+
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..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
@@ -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.
+ *
+ * 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 int getMaxConnections()
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.
+ *
+ * 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 String toString()
", 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 99b927efb2ba..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
@@ -21,15 +21,42 @@
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 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)
{
@@ -39,84 +66,144 @@ private static ObjectMapper mapperWithRuntimeInfo(RuntimeInfo runtimeInfo)
}
@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 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 properties, boolean expected)
+ {
+ Assertions.assertEquals(expected, bind(properties).isCrossRegionAccessEnabled());
+ }
+
+ private static Stream 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 c59f12670fd0..b60e80585e51 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 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
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 dd86e9e610de..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
@@ -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;
@@ -376,9 +377,11 @@ public static ServerSideEncryptingAmazonS3.Builder builder(
.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 static ServerSideEncryptingAmazonS3.Builder builder(
assumeRoleArn,
assumeRoleExternalId,
awsEndpointConfig,
- credentialsProvider
+ credentialsProvider,
+ awsClientConfig
);
}
@@ -425,6 +429,17 @@ public static ServerSideEncryptingAmazonS3.Builder builder(
.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 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());
@@ -459,6 +475,9 @@ public static AwsCredentialsProvider createAssumeRoleCredentialsProvider(
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 9840c4a68f97..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
@@ -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,10 @@ public void testS3InputSourceUseEndPointClientProxy()
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 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
{