This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi 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 d7207f5477c [FLINK-40387][s3] Add option to disable S3 SDK retry
circuit breaker
d7207f5477c is described below
commit d7207f5477c2712966975deda28c22b5f63ee4a5
Author: Gabor Somogyi <[email protected]>
AuthorDate: Tue Aug 18 14:18:34 2026 +0200
[FLINK-40387][s3] Add option to disable S3 SDK retry circuit breaker
---
.../flink/fs/s3native/NativeS3FileSystemFactory.java | 15 +++++++++++++++
.../org/apache/flink/fs/s3native/S3ClientProvider.java | 17 +++++++++++++++++
.../fs/s3native/NativeS3FileSystemFactoryTest.java | 13 +++++++++++++
.../apache/flink/fs/s3native/S3ClientProviderTest.java | 14 ++++++++++++++
4 files changed, 59 insertions(+)
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 cb4fd4788f5..23d6c22314c 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
@@ -282,6 +282,20 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
"Maximum delay cap for exponential backoff,
applied to both "
+ "normal and throttle retry paths.");
+ public static final ConfigOption<Boolean> RETRY_CIRCUIT_BREAKER_ENABLED =
+ ConfigOptions.key("s3.retry.circuit-breaker.enabled")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether the AWS SDK's retry circuit breaker
(token bucket) is enabled. "
+ + "The SDK stops retrying once a shared
per-client token bucket is "
+ + "drained by recent failures, regardless
of the retry/backoff settings "
+ + "above. Under a high volume of
concurrent requests hitting S3 "
+ + "throttling (e.g. large incremental
checkpoints), this bucket can drain "
+ + "within seconds, causing retries to be
abandoned well before the "
+ + "configured backoff and retry count are
exhausted. Disabled by default "
+ + "so the retry/backoff settings above
fully govern retry behavior.");
+
public static final ConfigOption<Duration> CONNECTION_TIMEOUT =
ConfigOptions.key("s3.connection.timeout")
.durationType()
@@ -615,6 +629,7 @@ public class NativeS3FileSystemFactory implements
FileSystemFactory {
.retryBaseDelay(config.get(RETRY_BASE_DELAY))
.retryThrottleBaseDelay(config.get(RETRY_THROTTLE_BASE_DELAY))
.retryMaxBackoff(config.get(RETRY_MAX_BACKOFF))
+
.retryCircuitBreakerEnabled(config.get(RETRY_CIRCUIT_BREAKER_ENABLED))
.credentialsProviderClasses(credentialsProviderClasses)
.encryptionConfig(encryptionConfig)
.useCrt(crtEnabled);
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 a12ef1da405..aa31db64696 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
@@ -97,6 +97,7 @@ class S3ClientProvider implements AutoCloseableAsync {
private final Duration retryBaseDelay;
private final Duration retryThrottleBaseDelay;
private final Duration retryMaxBackoff;
+ private final boolean retryCircuitBreakerEnabled;
@Nullable private final String region;
@Nullable private final String endpoint;
@Nullable private final String assumeRoleArn;
@@ -129,6 +130,7 @@ class S3ClientProvider implements AutoCloseableAsync {
Duration retryBaseDelay,
Duration retryThrottleBaseDelay,
Duration retryMaxBackoff,
+ boolean retryCircuitBreakerEnabled,
@Nullable String region,
@Nullable String endpoint,
@Nullable String assumeRoleArn,
@@ -172,6 +174,7 @@ class S3ClientProvider implements AutoCloseableAsync {
retryThrottleBaseDelay, "retryThrottleBaseDelay must
not be null");
this.retryMaxBackoff =
Preconditions.checkNotNull(retryMaxBackoff, "retryMaxBackoff
must not be null");
+ this.retryCircuitBreakerEnabled = retryCircuitBreakerEnabled;
this.region = region;
this.endpoint = endpoint;
this.assumeRoleArn = assumeRoleArn;
@@ -270,6 +273,11 @@ class S3ClientProvider implements AutoCloseableAsync {
return retryMaxBackoff;
}
+ @VisibleForTesting
+ boolean isRetryCircuitBreakerEnabled() {
+ return retryCircuitBreakerEnabled;
+ }
+
@VisibleForTesting
@Nullable
String getRegion() {
@@ -427,6 +435,8 @@ class S3ClientProvider implements AutoCloseableAsync {
NativeS3FileSystemFactory.RETRY_THROTTLE_BASE_DELAY.defaultValue();
private Duration retryMaxBackoff =
NativeS3FileSystemFactory.RETRY_MAX_BACKOFF.defaultValue();
+ private boolean retryCircuitBreakerEnabled =
+
NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED.defaultValue();
private Duration clientCloseTimeout =
NativeS3FileSystemFactory.CLIENT_CLOSE_TIMEOUT.defaultValue();
@@ -550,6 +560,11 @@ class S3ClientProvider implements AutoCloseableAsync {
return this;
}
+ public Builder retryCircuitBreakerEnabled(boolean
retryCircuitBreakerEnabled) {
+ this.retryCircuitBreakerEnabled = retryCircuitBreakerEnabled;
+ return this;
+ }
+
public Builder assumeRoleArn(@Nullable String assumeRoleArn) {
this.assumeRoleArn = assumeRoleArn;
return this;
@@ -671,6 +686,7 @@ class S3ClientProvider implements AutoCloseableAsync {
BackoffStrategy.exponentialDelay(
retryThrottleBaseDelay,
retryMaxBackoff))
+
.circuitBreakerEnabled(retryCircuitBreakerEnabled)
.build())
.build();
@@ -711,6 +727,7 @@ class S3ClientProvider implements AutoCloseableAsync {
retryBaseDelay,
retryThrottleBaseDelay,
retryMaxBackoff,
+ retryCircuitBreakerEnabled,
region,
endpoint,
assumeRoleArn,
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 bc6171919ae..902bc9ae9b6 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
@@ -359,6 +359,19 @@ class NativeS3FileSystemFactoryTest {
.isEqualTo(Duration.ofSeconds(30));
}
+ @Test
+ void testRetryCircuitBreakerEnabledDefault() throws Exception {
+
assertThat(createFs(baseConfig()).getClientProvider().isRetryCircuitBreakerEnabled())
+
.isEqualTo(NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED.defaultValue());
+ }
+
+ @Test
+ void testRetryCircuitBreakerEnabledExplicitlyConfigured() throws Exception
{
+ Configuration config = baseConfig();
+ config.set(NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED,
true);
+
assertThat(createFs(config).getClientProvider().isRetryCircuitBreakerEnabled()).isTrue();
+ }
+
// --- Timeouts ---
@Test
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 26985647165..935960837a3 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
@@ -237,6 +237,20 @@ class S3ClientProviderTest {
.isEqualTo(NativeS3FileSystemFactory.RETRY_THROTTLE_BASE_DELAY.defaultValue());
assertThat(provider.getRetryMaxBackoff())
.isEqualTo(NativeS3FileSystemFactory.RETRY_MAX_BACKOFF.defaultValue());
+ assertThat(provider.isRetryCircuitBreakerEnabled())
+
.isEqualTo(NativeS3FileSystemFactory.RETRY_CIRCUIT_BREAKER_ENABLED.defaultValue());
+ }
+
+ @Test
+ void testRetryCircuitBreakerEnabledOverride() {
+ S3ClientProvider provider =
+ S3ClientProvider.builder()
+ .endpoint(DUMMY_ENDPOINT)
+ .region(DUMMY_REGION)
+ .retryCircuitBreakerEnabled(true)
+ .build();
+
+ assertThat(provider.isRetryCircuitBreakerEnabled()).isTrue();
}
@Test
