This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 29655b36d7b [FLINK-41014][s3] Add an adaptive retry mode to the native
S3 connector
29655b36d7b is described below
commit 29655b36d7b64814df010c96b781060c97ad0c82
Author: Gabor Somogyi <[email protected]>
AuthorDate: Tue Nov 3 18:39:26 2026 +0100
[FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector
* [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector
* [FLINK-41014][s3] Upgrade the AWS SDK of the native S3 connector to 2.57.2
The adaptive retry mode relies on the SDK's AdaptiveRetryStrategy, whose
client-side rate limiting was too aggressive (the allowed request rate
could end up far below what the object store serves) and unstable (it
kept varying although the server side rate was constant). That is fixed
by https://github.com/aws/aws-sdk-java-v2/pull/7197, first released in
SDK 2.52.0, so the SDK is moved to the latest release, 2.57.2.
- aws-crt is moved to 0.48.5, the version SDK 2.57.2 is built against.
download-crt-jars.sh resolves it from the SDK version and needs no
change apart from dropping an outdated example version.
- Since SDK 2.46.0 every service client pulls in apache5-client, which
becomes the default HTTP client of clients built without an explicit
one (e.g. the STS client of the web identity credentials provider) and
would add org.apache.hc.* classes that the shade configuration does
not relocate. It is excluded to keep the Apache 4 based HTTP stack.
Moving to apache5-client, as apache-client is deprecated since SDK
2.54.15, is left for later.
- NOTICE: the SDK versions are updated, and software.amazon.eventstream
is replaced by software.amazon.awssdk:eventstream because the SDK
dropped the former in 2.57.0 and ships its own module instead.
Generated-by: Claude Code 2.1.293 (Claude Sonnet 5.5)
* [FLINK-41014][s3] Address review feedback on the adaptive retry mode
- Shorten the description of s3.retry.mode and link the AWS SDK retry
guide. The README now says that the rate limiter exists per bucket and
TaskManager, as the factory builds one client provider per bucket.
- Log the resolved retry mode at INFO when the client provider is built.
- Test that the client of the provider is built with the strategy of the
configured mode and the configured number of attempts. The getters
checked so far did not notice a provider that ignores the mode.
- Assert the attempts of a failed request with S3Exception#numAttempts
instead of the SDK's message, and drop the assertion on the message of
the circuit breaker, as the SDK may reword both.
Generated-by: Claude Code 2.1.293 (Claude Sonnet 5.5)
---
flink-filesystems/flink-s3-fs-native/README.md | 3 +-
flink-filesystems/flink-s3-fs-native/pom.xml | 27 +++-
.../fs/s3native/NativeS3FileSystemFactory.java | 22 ++++
.../apache/flink/fs/s3native/S3ClientProvider.java | 41 ++++--
.../flink/fs/s3native/S3RetryStrategyFactory.java | 90 +++++++++++++
.../src/main/resources/META-INF/NOTICE | 68 +++++-----
.../fs/s3native/NativeS3FileSystemFactoryTest.java | 23 ++++
.../fs/s3native/S3ClientProviderRetryModeTest.java | 140 +++++++++++++++++++++
.../flink/fs/s3native/S3ClientProviderTest.java | 60 +++++++++
.../fs/s3native/S3RetryStrategyFactoryTest.java | 121 ++++++++++++++++++
.../flink-s3-fs-native/tools/download-crt-jars.sh | 2 +-
11 files changed, 546 insertions(+), 51 deletions(-)
diff --git a/flink-filesystems/flink-s3-fs-native/README.md
b/flink-filesystems/flink-s3-fs-native/README.md
index 4686186099e..c8c67493bdf 100644
--- a/flink-filesystems/flink-s3-fs-native/README.md
+++ b/flink-filesystems/flink-s3-fs-native/README.md
@@ -116,6 +116,7 @@ The plugin emits the following metric names:
| Key | Default | Description |
|-----|---------|-------------|
| s3.retry.max-num-retries | 3 | Maximum retry attempts for failed S3
requests. Uses the AWS SDK's default retry strategy (exponential backoff with
jitter). Set to `0` to disable retries |
+| s3.retry.mode | standard | `standard`: every request retries on its own,
which keeps all of them retrying at full speed against a request limit.
`adaptive`: additionally a client-side rate limiter, shared by all requests of
the S3 client (one per bucket and TaskManager), slows the requests down after
throttling answers (HTTP 429, S3 `SlowDown`) and lets them speed up after
successful ones, so that many parallel writers settle at the request rate the
object store accepts |
| s3.connection.timeout | 60s | HTTP connection timeout for the S3 client
(time to establish a connection) |
| s3.socket.timeout | 60s | HTTP socket timeout for the S3 client (time to
wait for data after connection is established) |
| s3.connection.max-idle-time | 60s | Maximum idle time for HTTP connections
in the connection pool |
@@ -443,7 +444,7 @@ If `mvn` is unavailable, fetch the JAR by hand from Maven
Central:
`aws-crt-<version>.jar` (groupId: `software.amazon.awssdk.crt`) — the version
is
declared as the `<dependency>` on `software.amazon.awssdk.crt:aws-crt` inside
`aws-crt-client-<version>.pom` on Maven Central. The `aws-crt` artifact uses an
-independent versioning scheme (e.g. `0.45.x`) that does **not** track the AWS
SDK
+independent versioning scheme (e.g. `0.48.x`) that does **not** track the AWS
SDK
version.
`aws-crt-client` does **not** need to be downloaded — it is bundled inside the
fat JAR.
diff --git a/flink-filesystems/flink-s3-fs-native/pom.xml
b/flink-filesystems/flink-s3-fs-native/pom.xml
index 1bdd1576b60..0e53bf8588c 100644
--- a/flink-filesystems/flink-s3-fs-native/pom.xml
+++ b/flink-filesystems/flink-s3-fs-native/pom.xml
@@ -31,11 +31,11 @@ under the License.
<packaging>jar</packaging>
<properties>
- <fs.s3.aws.sdk.version>2.44.4</fs.s3.aws.sdk.version>
+ <fs.s3.aws.sdk.version>2.57.2</fs.s3.aws.sdk.version>
<!-- aws-crt uses a separate versioning scheme from the AWS
SDK; this version was
validated against aws-crt-client ${fs.s3.aws.sdk.version}.
Update when bumping
the SDK version (check aws-crt-client-<version>.pom →
awscrt.version). -->
- <fs.s3.aws.crt.version>0.45.1</fs.s3.aws.crt.version>
+ <fs.s3.aws.crt.version>0.48.5</fs.s3.aws.crt.version>
<japicmp.skip>true</japicmp.skip>
<surefire.module.config>--add-opens=java.base/java.util=ALL-UNNAMED</surefire.module.config>
</properties>
@@ -55,11 +55,22 @@ under the License.
<scope>test</scope>
</dependency>
+ <!-- Since SDK 2.46.0 every service client pulls in
apache5-client at runtime, which makes it
+ the default HTTP client of clients built without an
explicit one (e.g. the STS client
+ of the web identity credentials provider) and bundles
org.apache.hc.*, which is not
+ relocated below. The connector uses apache-client
explicitly, so keep the HTTP stack
+ as it was and exclude apache5-client from every dependency
that brings it. -->
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>s3</artifactId>
<version>${fs.s3.aws.sdk.version}</version>
<optional>${flink.markBundledAsOptional}</optional>
+ <exclusions>
+ <exclusion>
+
<groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<!-- S3 Transfer Manager for async/high-throughput transfers -->
@@ -68,6 +79,12 @@ under the License.
<artifactId>s3-transfer-manager</artifactId>
<version>${fs.s3.aws.sdk.version}</version>
<optional>${flink.markBundledAsOptional}</optional>
+ <exclusions>
+ <exclusion>
+
<groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
@@ -75,6 +92,12 @@ under the License.
<artifactId>sts</artifactId>
<version>${fs.s3.aws.sdk.version}</version>
<optional>${flink.markBundledAsOptional}</optional>
+ <exclusions>
+ <exclusion>
+
<groupId>software.amazon.awssdk</groupId>
+ <artifactId>apache5-client</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<!-- Netty async HTTP client (required for Transfer Manager) -->
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
index 55a48debfda..72639c145bd 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java
@@ -312,6 +312,27 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
+ "configured backoff and retry count are
exhausted. Disabled by default "
+ "so the retry/backoff settings above
fully govern retry behavior.");
+ /** How the AWS SDK clients retry failed requests, see {@link
#RETRY_MODE}. */
+ public enum RetryMode {
+ STANDARD,
+ ADAPTIVE
+ }
+
+ public static final ConfigOption<RetryMode> RETRY_MODE =
+ ConfigOptions.key("s3.retry.mode")
+ .enumType(RetryMode.class)
+ .defaultValue(RetryMode.STANDARD)
+ .withDescription(
+ "How the AWS SDK retries failed requests.
'standard' (default): every "
+ + "request retries on its own, with the
delays set by the "
+ + "s3.retry.* options. 'adaptive': the
same delays, plus a "
+ + "client-side rate limiter that slows
requests down after "
+ + "throttling answers (HTTP 429, S3
SlowDown) and can delay a "
+ + "request before its first attempt. It
trades latency for "
+ + "success rate. The AWS CRT async client
(s3.crt.enabled) has "
+ + "retries of its own and ignores this
option. See "
+ +
"https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/retry-strategy.html";);
+
public static final ConfigOption<Duration> CONNECTION_TIMEOUT =
ConfigOptions.key("s3.connection.timeout")
.durationType()
@@ -722,6 +743,7 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory, MetricsAwar
.retryThrottleBaseDelay(config.get(RETRY_THROTTLE_BASE_DELAY))
.retryMaxBackoff(config.get(RETRY_MAX_BACKOFF))
.retryCircuitBreakerEnabled(config.get(RETRY_CIRCUIT_BREAKER_ENABLED))
+ .retryMode(config.get(RETRY_MODE))
.credentialsProviderClasses(credentialsProviderClasses)
.encryptionConfig(encryptionConfig)
.metricPublisher(resolveMetricBridge(config))
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
index 634409da4d3..fe8153fd549 100644
---
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3ClientProvider.java
@@ -39,8 +39,6 @@ import
software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
import software.amazon.awssdk.metrics.MetricPublisher;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain;
-import software.amazon.awssdk.retries.StandardRetryStrategy;
-import software.amazon.awssdk.retries.api.BackoffStrategy;
import software.amazon.awssdk.services.s3.S3AsyncClient;
import software.amazon.awssdk.services.s3.S3AsyncClientBuilder;
import software.amazon.awssdk.services.s3.S3Client;
@@ -99,6 +97,7 @@ class S3ClientProvider implements AutoCloseableAsync {
private final Duration retryThrottleBaseDelay;
private final Duration retryMaxBackoff;
private final boolean retryCircuitBreakerEnabled;
+ private final NativeS3FileSystemFactory.RetryMode retryMode;
@Nullable private final String region;
@Nullable private final String endpoint;
@Nullable private final String assumeRoleArn;
@@ -132,6 +131,7 @@ class S3ClientProvider implements AutoCloseableAsync {
Duration retryThrottleBaseDelay,
Duration retryMaxBackoff,
boolean retryCircuitBreakerEnabled,
+ NativeS3FileSystemFactory.RetryMode retryMode,
@Nullable String region,
@Nullable String endpoint,
@Nullable String assumeRoleArn,
@@ -176,6 +176,7 @@ class S3ClientProvider implements AutoCloseableAsync {
this.retryMaxBackoff =
Preconditions.checkNotNull(retryMaxBackoff, "retryMaxBackoff
must not be null");
this.retryCircuitBreakerEnabled = retryCircuitBreakerEnabled;
+ this.retryMode = Preconditions.checkNotNull(retryMode, "retryMode must
not be null");
this.region = region;
this.endpoint = endpoint;
this.assumeRoleArn = assumeRoleArn;
@@ -279,6 +280,11 @@ class S3ClientProvider implements AutoCloseableAsync {
return retryCircuitBreakerEnabled;
}
+ @VisibleForTesting
+ NativeS3FileSystemFactory.RetryMode getRetryMode() {
+ return retryMode;
+ }
+
@VisibleForTesting
@Nullable
String getRegion() {
@@ -438,6 +444,8 @@ class S3ClientProvider implements AutoCloseableAsync {
NativeS3FileSystemFactory.RETRY_MAX_BACKOFF.defaultValue();
private boolean retryCircuitBreakerEnabled =
NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED.defaultValue();
+ private NativeS3FileSystemFactory.RetryMode retryMode =
+ NativeS3FileSystemFactory.RETRY_MODE.defaultValue();
private Duration clientCloseTimeout =
NativeS3FileSystemFactory.CLIENT_CLOSE_TIMEOUT.defaultValue();
@@ -570,6 +578,11 @@ class S3ClientProvider implements AutoCloseableAsync {
return this;
}
+ public Builder retryMode(NativeS3FileSystemFactory.RetryMode
retryMode) {
+ this.retryMode = Preconditions.checkNotNull(retryMode, "retryMode
must not be null");
+ return this;
+ }
+
public Builder assumeRoleArn(@Nullable String assumeRoleArn) {
this.assumeRoleArn = assumeRoleArn;
return this;
@@ -687,17 +700,18 @@ class S3ClientProvider implements AutoCloseableAsync {
ClientOverrideConfiguration.Builder overrideConfigBuilder =
ClientOverrideConfiguration.builder()
.retryStrategy(
- StandardRetryStrategy.builder()
- .maxAttempts(maxRetries + 1)
- .backoffStrategy(
-
BackoffStrategy.exponentialDelay(
- retryBaseDelay,
retryMaxBackoff))
- .throttlingBackoffStrategy(
-
BackoffStrategy.exponentialDelay(
-
retryThrottleBaseDelay,
- retryMaxBackoff))
-
.circuitBreakerEnabled(retryCircuitBreakerEnabled)
- .build());
+ S3RetryStrategyFactory.create(
+ retryMode,
+ maxRetries,
+ retryBaseDelay,
+ retryThrottleBaseDelay,
+ retryMaxBackoff,
+ retryCircuitBreakerEnabled));
+
+ LOG.info(
+ "S3 retry mode {} (s3.retry.mode) with up to {} retries
per request",
+ retryMode,
+ maxRetries);
if (useCrt) {
LOG.info(
@@ -742,6 +756,7 @@ class S3ClientProvider implements AutoCloseableAsync {
retryThrottleBaseDelay,
retryMaxBackoff,
retryCircuitBreakerEnabled,
+ retryMode,
region,
endpoint,
assumeRoleArn,
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3RetryStrategyFactory.java
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3RetryStrategyFactory.java
new file mode 100644
index 00000000000..c911634b7fb
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/S3RetryStrategyFactory.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native;
+
+import org.apache.flink.annotation.Internal;
+
+import software.amazon.awssdk.retries.AdaptiveRetryStrategy;
+import software.amazon.awssdk.retries.StandardRetryStrategy;
+import software.amazon.awssdk.retries.api.BackoffStrategy;
+import software.amazon.awssdk.retries.api.RetryStrategy;
+import software.amazon.awssdk.retries.internal.DefaultAdaptiveRetryStrategy;
+
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * Builds the AWS SDK {@link RetryStrategy} used by the native S3 client's
sync/async clients: the
+ * standard or the adaptive one, see {@link
NativeS3FileSystemFactory#RETRY_MODE}. The delays are
+ * validated by {@link S3ClientProvider.Builder}.
+ */
+@Internal
+final class S3RetryStrategyFactory {
+
+ private S3RetryStrategyFactory() {}
+
+ static RetryStrategy create(
+ NativeS3FileSystemFactory.RetryMode retryMode,
+ int maxRetries,
+ Duration retryBaseDelay,
+ Duration retryThrottleBaseDelay,
+ Duration retryMaxBackoff,
+ boolean retryCircuitBreakerEnabled) {
+ BackoffStrategy backoff =
BackoffStrategy.exponentialDelay(retryBaseDelay, retryMaxBackoff);
+ BackoffStrategy throttlingBackoff =
+ BackoffStrategy.exponentialDelay(retryThrottleBaseDelay,
retryMaxBackoff);
+
+ switch (retryMode) {
+ case STANDARD:
+ return StandardRetryStrategy.builder()
+ .maxAttempts(maxRetries + 1)
+ .backoffStrategy(backoff)
+ .throttlingBackoffStrategy(throttlingBackoff)
+ .circuitBreakerEnabled(retryCircuitBreakerEnabled)
+ .build();
+ case ADAPTIVE:
+ return createAdaptive(
+ maxRetries, backoff, throttlingBackoff,
retryCircuitBreakerEnabled);
+ default:
+ throw new IllegalArgumentException(
+ "Unknown retry mode: "
+ + retryMode
+ + ". Supported values: "
+ +
Arrays.toString(NativeS3FileSystemFactory.RetryMode.values()));
+ }
+ }
+
+ private static RetryStrategy createAdaptive(
+ int maxRetries,
+ BackoffStrategy backoff,
+ BackoffStrategy throttlingBackoff,
+ boolean retryCircuitBreakerEnabled) {
+ AdaptiveRetryStrategy.Builder builder =
+ AdaptiveRetryStrategy.builder()
+ .maxAttempts(maxRetries + 1)
+ .backoffStrategy(backoff)
+ .throttlingBackoffStrategy(throttlingBackoff);
+ // The public builder of the adaptive strategy has no switch for the
circuit breaker, so it
+ // would stay on and a throttling storm would drain its token bucket
within seconds (see
+ // RETRY_CIRCUIT_BREAKER_ENABLED). The SDK's own builder has the
switch.
+ ((DefaultAdaptiveRetryStrategy.Builder) builder)
+ .circuitBreakerEnabled(retryCircuitBreakerEnabled);
+ return builder.build();
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
b/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
index 0f263329679..e61f5386348 100644
--- a/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
+++ b/flink-filesystems/flink-s3-fs-native/src/main/resources/META-INF/NOTICE
@@ -6,40 +6,40 @@ The Apache Software Foundation (http://www.apache.org/).
This project bundles the following dependencies under the Apache Software
License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
-- software.amazon.awssdk:annotations:2.44.4
-- software.amazon.awssdk:apache-client:2.44.4
-- software.amazon.awssdk:arns:2.44.4
-- software.amazon.awssdk:auth:2.44.4
-- software.amazon.awssdk:aws-core:2.44.4
-- software.amazon.awssdk:aws-crt-client:2.44.4
-- software.amazon.awssdk:aws-query-protocol:2.44.4
-- software.amazon.awssdk:aws-xml-protocol:2.44.4
-- software.amazon.awssdk:checksums:2.44.4
-- software.amazon.awssdk:checksums-spi:2.44.4
-- software.amazon.awssdk:crt-core:2.44.4
-- software.amazon.awssdk:endpoints-spi:2.44.4
-- software.amazon.awssdk:http-auth:2.44.4
-- software.amazon.awssdk:http-auth-aws:2.44.4
-- software.amazon.awssdk:http-auth-aws-eventstream:2.44.4
-- software.amazon.awssdk:http-auth-spi:2.44.4
-- software.amazon.awssdk:http-client-spi:2.44.4
-- software.amazon.awssdk:identity-spi:2.44.4
-- software.amazon.awssdk:json-utils:2.44.4
-- software.amazon.awssdk:metrics-spi:2.44.4
-- software.amazon.awssdk:netty-nio-client:2.44.4
-- software.amazon.awssdk:profiles:2.44.4
-- software.amazon.awssdk:protocol-core:2.44.4
-- software.amazon.awssdk:regions:2.44.4
-- software.amazon.awssdk:retries:2.44.4
-- software.amazon.awssdk:retries-spi:2.44.4
-- software.amazon.awssdk:s3:2.44.4
-- software.amazon.awssdk:s3-transfer-manager:2.44.4
-- software.amazon.awssdk:sdk-core:2.44.4
-- software.amazon.awssdk:sts:2.44.4
-- software.amazon.awssdk:third-party-jackson-core:2.44.4
-- software.amazon.awssdk:utils:2.44.4
-- software.amazon.awssdk:utils-lite:2.44.4
-- software.amazon.eventstream:eventstream:1.0.1
+- software.amazon.awssdk:annotations:2.57.2
+- software.amazon.awssdk:apache-client:2.57.2
+- software.amazon.awssdk:arns:2.57.2
+- software.amazon.awssdk:auth:2.57.2
+- software.amazon.awssdk:aws-core:2.57.2
+- software.amazon.awssdk:aws-crt-client:2.57.2
+- software.amazon.awssdk:aws-query-protocol:2.57.2
+- software.amazon.awssdk:aws-xml-protocol:2.57.2
+- software.amazon.awssdk:checksums:2.57.2
+- software.amazon.awssdk:checksums-spi:2.57.2
+- software.amazon.awssdk:crt-core:2.57.2
+- software.amazon.awssdk:endpoints-spi:2.57.2
+- software.amazon.awssdk:eventstream:2.57.2
+- software.amazon.awssdk:http-auth:2.57.2
+- software.amazon.awssdk:http-auth-aws:2.57.2
+- software.amazon.awssdk:http-auth-aws-eventstream:2.57.2
+- software.amazon.awssdk:http-auth-spi:2.57.2
+- software.amazon.awssdk:http-client-spi:2.57.2
+- software.amazon.awssdk:identity-spi:2.57.2
+- software.amazon.awssdk:json-utils:2.57.2
+- software.amazon.awssdk:metrics-spi:2.57.2
+- software.amazon.awssdk:netty-nio-client:2.57.2
+- software.amazon.awssdk:profiles:2.57.2
+- software.amazon.awssdk:protocol-core:2.57.2
+- software.amazon.awssdk:regions:2.57.2
+- software.amazon.awssdk:retries:2.57.2
+- software.amazon.awssdk:retries-spi:2.57.2
+- software.amazon.awssdk:s3:2.57.2
+- software.amazon.awssdk:s3-transfer-manager:2.57.2
+- software.amazon.awssdk:sdk-core:2.57.2
+- software.amazon.awssdk:sts:2.57.2
+- software.amazon.awssdk:third-party-jackson-core:2.57.2
+- software.amazon.awssdk:utils:2.57.2
+- software.amazon.awssdk:utils-lite:2.57.2
- org.apache.httpcomponents:httpclient:4.5.13
- org.apache.httpcomponents:httpcore:4.4.14
- commons-codec:commons-codec:1.15
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
index 2d927d4fde7..9e208fff4b2 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java
@@ -439,6 +439,29 @@ class NativeS3FileSystemFactoryTest {
assertThat(createFs(config).getClientProvider().isRetryCircuitBreakerEnabled()).isTrue();
}
+ @Test
+ void testRetryModeDefault() throws Exception {
+ assertThat(createFs(baseConfig()).getClientProvider().getRetryMode())
+ .isEqualTo(NativeS3FileSystemFactory.RetryMode.STANDARD);
+ }
+
+ @Test
+ void testRetryModeExplicitlyConfigured() throws Exception {
+ Configuration config = baseConfig();
+ config.set(
+ NativeS3FileSystemFactory.RETRY_MODE,
NativeS3FileSystemFactory.RetryMode.ADAPTIVE);
+ assertThat(createFs(config).getClientProvider().getRetryMode())
+ .isEqualTo(NativeS3FileSystemFactory.RetryMode.ADAPTIVE);
+ }
+
+ @Test
+ void testRetryModeIsReadFromTheLowerCaseValueOfAFlinkConfigFile() throws
Exception {
+ Configuration config = baseConfig();
+ config.setString("s3.retry.mode", "adaptive");
+ assertThat(createFs(config).getClientProvider().getRetryMode())
+ .isEqualTo(NativeS3FileSystemFactory.RetryMode.ADAPTIVE);
+ }
+
// --- Timeouts ---
@Test
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderRetryModeTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderRetryModeTest.java
new file mode 100644
index 00000000000..840608f4c65
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderRetryModeTest.java
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native;
+
+import org.apache.flink.fs.s3native.NativeS3FileSystemFactory.RetryMode;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that a client built for either {@link RetryMode} retries what an
object store answers with
+ * SlowDown, against a local server that throttles a number of requests and
then answers.
+ */
+class S3ClientProviderRetryModeTest {
+
+ private static final String SLOW_DOWN =
+ "<Error><Code>SlowDown</Code><Message>Please reduce your request
rate.</Message>"
+ + "<RequestId>1</RequestId></Error>";
+
+ private final AtomicInteger requests = new AtomicInteger();
+
+ private HttpServer server;
+ private int throttledRequests;
+ private S3ClientProvider provider;
+
+ @BeforeEach
+ void startServer() throws IOException {
+ server = HttpServer.create(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+ server.createContext(
+ "/",
+ exchange -> {
+ if (requests.incrementAndGet() <= throttledRequests) {
+ respond(exchange, 503, SLOW_DOWN);
+ } else {
+ respond(exchange, 200, "hello");
+ }
+ });
+ server.start();
+ }
+
+ @AfterEach
+ void stopServer() throws Exception {
+ if (provider != null) {
+ provider.close();
+ }
+ server.stop(0);
+ }
+
+ @ParameterizedTest
+ @EnumSource(RetryMode.class)
+ void testAThrottledRequestIsRetriedUntilTheStoreAnswers(RetryMode mode) {
+ throttledRequests = 1;
+ provider = provider(mode, 5);
+
+ String body = getObject().asUtf8String();
+
+ assertThat(body).isEqualTo("hello");
+ assertThat(requests).hasValue(2);
+ }
+
+ @ParameterizedTest
+ @EnumSource(RetryMode.class)
+ void
testARequestThatStaysThrottledFailsAfterTheConfiguredRetries(RetryMode mode) {
+ throttledRequests = Integer.MAX_VALUE;
+ provider = provider(mode, 1);
+
+ assertThatThrownBy(this::getObject)
+ .isInstanceOfSatisfying(
+ S3Exception.class,
+ e -> {
+ assertThat(e.statusCode()).isEqualTo(503);
+ assertThat(e.numAttempts()).isEqualTo(2);
+ });
+ assertThat(requests).hasValue(2);
+ }
+
+ private ResponseBytes<GetObjectResponse> getObject() {
+ return provider.getS3Client()
+
.getObjectAsBytes(GetObjectRequest.builder().bucket("bucket").key("key").build());
+ }
+
+ private S3ClientProvider provider(RetryMode mode, int maxRetries) {
+ return S3ClientProvider.builder()
+ .endpoint("http://127.0.0.1:"; + server.getAddress().getPort())
+ .region("us-east-1")
+ .pathStyleAccess(true)
+ .accessKey("access")
+ .secretKey("secret")
+ .retryMode(mode)
+ .maxRetries(maxRetries)
+ .retryBaseDelay(Duration.ofMillis(1))
+ .retryThrottleBaseDelay(Duration.ofMillis(1))
+ .retryMaxBackoff(Duration.ofMillis(10))
+ .build();
+ }
+
+ private static void respond(HttpExchange exchange, int status, String
body) throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/xml");
+ exchange.sendResponseHeaders(status, bytes.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(bytes);
+ }
+ }
+}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
index 935960837a3..eeec85d641b 100644
---
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderTest.java
@@ -28,6 +28,9 @@ import software.amazon.awssdk.auth.credentials.AwsCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProviderChain;
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.retries.AdaptiveRetryStrategy;
+import software.amazon.awssdk.retries.StandardRetryStrategy;
+import software.amazon.awssdk.retries.api.RetryStrategy;
import
software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
import java.lang.reflect.Field;
@@ -239,6 +242,54 @@ class S3ClientProviderTest {
.isEqualTo(NativeS3FileSystemFactory.RETRY_MAX_BACKOFF.defaultValue());
assertThat(provider.isRetryCircuitBreakerEnabled())
.isEqualTo(NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED.defaultValue());
+ assertThat(provider.getRetryMode())
+
.isEqualTo(NativeS3FileSystemFactory.RETRY_MODE.defaultValue());
+ }
+
+ @Test
+ void testRetryModeOverride() {
+ S3ClientProvider provider =
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+
.retryMode(NativeS3FileSystemFactory.RetryMode.ADAPTIVE)
+ .build();
+
+
assertThat(provider.getRetryMode()).isEqualTo(NativeS3FileSystemFactory.RetryMode.ADAPTIVE);
+ }
+
+ @Test
+ void testStandardModeBuildsAClientWithTheStandardStrategy() {
+ S3ClientProvider provider =
+ track(
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+
.retryMode(NativeS3FileSystemFactory.RetryMode.STANDARD)
+ .maxRetries(5)
+ .build());
+
+ RetryStrategy strategy = retryStrategyOf(provider);
+
+ assertThat(strategy).isInstanceOf(StandardRetryStrategy.class);
+ assertThat(strategy.maxAttempts()).isEqualTo(6);
+ }
+
+ @Test
+ void testAdaptiveModeBuildsAClientWithTheAdaptiveStrategy() {
+ S3ClientProvider provider =
+ track(
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+
.retryMode(NativeS3FileSystemFactory.RetryMode.ADAPTIVE)
+ .maxRetries(5)
+ .build());
+
+ RetryStrategy strategy = retryStrategyOf(provider);
+
+ assertThat(strategy).isInstanceOf(AdaptiveRetryStrategy.class);
+ assertThat(strategy.maxAttempts()).isEqualTo(6);
}
@Test
@@ -382,4 +433,13 @@ class S3ClientProviderTest {
field.setAccessible(true);
return (List<AwsCredentialsProvider>) field.get(provider);
}
+
+ /** The retry strategy the SDK client of the provider was built with. */
+ private static RetryStrategy retryStrategyOf(S3ClientProvider provider) {
+ return provider.getS3Client()
+ .serviceClientConfiguration()
+ .overrideConfiguration()
+ .retryStrategy()
+ .orElseThrow(() -> new AssertionError("The client has no retry
strategy"));
+ }
}
diff --git
a/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3RetryStrategyFactoryTest.java
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3RetryStrategyFactoryTest.java
new file mode 100644
index 00000000000..eae16eda5ee
--- /dev/null
+++
b/flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3RetryStrategyFactoryTest.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.fs.s3native;
+
+import org.apache.flink.fs.s3native.NativeS3FileSystemFactory.RetryMode;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.retries.AdaptiveRetryStrategy;
+import software.amazon.awssdk.retries.StandardRetryStrategy;
+import software.amazon.awssdk.retries.api.AcquireInitialTokenRequest;
+import software.amazon.awssdk.retries.api.RefreshRetryTokenRequest;
+import software.amazon.awssdk.retries.api.RetryStrategy;
+import software.amazon.awssdk.retries.api.RetryToken;
+import software.amazon.awssdk.retries.api.TokenAcquisitionFailedException;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link S3RetryStrategyFactory}. */
+class S3RetryStrategyFactoryTest {
+
+ private static RetryStrategy create(RetryMode mode, boolean
circuitBreakerEnabled) {
+ return S3RetryStrategyFactory.create(
+ mode,
+ 10,
+ Duration.ofMillis(100),
+ Duration.ofSeconds(1),
+ Duration.ofSeconds(20),
+ circuitBreakerEnabled);
+ }
+
+ /**
+ * The strategy as the SDK client completes it: the client adds the
conditions that say what is
+ * retried, here everything.
+ */
+ private static RetryStrategy retryingEverything(RetryStrategy strategy) {
+ RetryStrategy.Builder<?, ?> builder = strategy.toBuilder();
+ builder.retryOnException(failure -> true);
+ return builder.build();
+ }
+
+ /** One call whose attempts fail up to the last one: all ten retries of
the strategy. */
+ private static void retryTenTimes(RetryStrategy strategy) {
+ SdkClientException failure = SdkClientException.create("boom");
+ RetryToken token =
+
strategy.acquireInitialToken(AcquireInitialTokenRequest.create("test")).token();
+ for (int retry = 0; retry < 10; retry++) {
+ token =
+ strategy.refreshRetryToken(
+ RefreshRetryTokenRequest.builder()
+ .token(token)
+ .failure(failure)
+ .build())
+ .token();
+ }
+ }
+
+ @Test
+ void testStandardModeBuildsTheStandardStrategy() {
+ assertThat(create(RetryMode.STANDARD,
false)).isInstanceOf(StandardRetryStrategy.class);
+ }
+
+ @Test
+ void testAdaptiveModeBuildsTheAdaptiveStrategy() {
+ assertThat(create(RetryMode.ADAPTIVE,
false)).isInstanceOf(AdaptiveRetryStrategy.class);
+ }
+
+ @ParameterizedTest
+ @EnumSource(RetryMode.class)
+ void testMaxAttemptsIsMaxRetriesPlusOne(RetryMode mode) {
+ assertThat(create(mode, false).maxAttempts()).isEqualTo(11);
+ }
+
+ @ParameterizedTest
+ @EnumSource(RetryMode.class)
+ void testCircuitBreakerStaysOffUnlessEnabled(RetryMode mode) {
+ RetryStrategy strategy = retryingEverything(create(mode, false));
+
+ // Ten retries cost 50 tokens of the 500 of the breaker: it would be
empty after ten calls.
+ for (int call = 0; call < 30; call++) {
+ retryTenTimes(strategy);
+ }
+ }
+
+ @ParameterizedTest
+ @EnumSource(RetryMode.class)
+ void testCircuitBreakerStopsTheRetriesWhenEnabled(RetryMode mode) {
+ RetryStrategy strategy = retryingEverything(create(mode, true));
+
+ // A call never exceeds its attempts (ten retries of the ten allowed)
and the same calls
+ // pass with the breaker off, so only the drained breaker can fail
them.
+ assertThatThrownBy(
+ () -> {
+ for (int call = 0; call < 30; call++) {
+ retryTenTimes(strategy);
+ }
+ })
+ .isInstanceOf(TokenAcquisitionFailedException.class);
+ }
+}
diff --git a/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
b/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
index d3e29fbe708..a0974bb2309 100755
--- a/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
+++ b/flink-filesystems/flink-s3-fs-native/tools/download-crt-jars.sh
@@ -110,7 +110,7 @@ CRT_VERSION="$(resolve_crt_version || true)"
if [[ -z "${CRT_VERSION}" ]]; then
echo "ERROR: aws-crt version could not be resolved for
aws-crt-client:${SDK_VERSION}." >&2
echo " Set AWS_CRT_VERSION explicitly and re-run, e.g." >&2
- echo " AWS_CRT_VERSION=0.33.6 $0 ${OUTPUT_DIR}" >&2
+ echo " AWS_CRT_VERSION=<version> $0 ${OUTPUT_DIR}" >&2
exit 1
fi
