This is an automated email from the ASF dual-hosted git repository.
Samrat002 pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 450d92b02e8 [FLINK-41014][s3] Add an adaptive retry mode to the native
S3 connector
450d92b02e8 is described below
commit 450d92b02e8778fdcbc14d159b8de9afbf91085d
Author: Gabor Somogyi <[email protected]>
AuthorDate: Wed Nov 4 15:12:06 2026 +0100
[FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector
---
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 37be4e53a75..e8bdb30a1f9 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
