You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "gaborgsomogyi (via GitHub)" <gi...@apache.org> on 2026/10/16 13:32:18 UTC
[PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
gaborgsomogyi opened a new pull request, #29499:
URL: https://github.com/apache/flink/pull/29499
## What is the purpose of the change
S3-compatible object stores, and S3 itself per prefix, limit the request rate and shed load with
`503 SlowDown` ("Please reduce your request rate") answers; some also close the connection. A sink with
many parallel writers runs into that limit easily, because every writer uploads its multipart parts
synchronously.
The native S3 connector retries with the AWS SDK's standard strategy, in which every request retries on
its own. Once the limit is reached, all writers back off for similar, short times and come back together,
so the request rate stays above the limit. A request that meets the limit on every attempt runs out of
attempts, the write fails, and with it the job. More retries do not fix that at higher parallelism. On an
object store with a request-rate quota, a 1 TiB streaming write failed in all three runs with the standard
strategy, each time on a single `UploadPart` request whose attempts all ended in a 503 answer or in a
connection closed by the store:
| Writers | `s3.retry.max-num-retries` | Failed after |
|--------:|---------------------------:|-------------:|
| 256 | 10 | 31 s |
| 256 | 20 | 2 min 20 s |
| 32 | 10 | ~5 min |
This PR adds the option `s3.retry.mode`:
- `standard`: the behavior so far.
- `adaptive` (the new default): the SDK's adaptive retry strategy. It makes the same attempts with the same
delays as `standard`, and adds a client-side rate limiter that is shared by all requests of an S3 client
(one per TaskManager). The limiter does nothing until the first throttling answer (HTTP 429 or S3
`SlowDown`). From then on, throttling answers reduce its send rate and successful answers let it recover,
so the writers settle at the rate the store accepts instead of all retrying at full speed. A request may
wait before its first attempt: the mode trades latency for success.
The circuit-breaker setting (`s3.retry.circuit-breaker.enabled`, off by default) applies to both modes.
Because the limiter stays out of the way until the store throttles, a deployment whose store does not
throttle is not affected by the new default. Where the store does throttle, requests are slowed down instead
of failing. `s3.retry.mode: standard` restores the previous behavior.
## Brief change log
- `NativeS3FileSystemFactory`: the option `s3.retry.mode` (`adaptive` | `standard`) with the nested enum
`RetryMode`, passed on to the client provider.
- `S3RetryStrategyFactory` (new, `@Internal`): creates the SDK retry strategy of a mode, with a `switch`
that fails on an unknown mode. Both modes use the same number of attempts, the same delays and the same
circuit-breaker setting.
- `S3ClientProvider`: carries the mode and takes the strategy from the factory, for the synchronous and
the asynchronous client.
- Module README: the option is added to the "Retries and Timeouts" table.
## Verifying this change
This change added tests and can be verified as follows:
- `S3RetryStrategyFactoryTest` (new): the strategy type of each mode; the number of attempts is the
number of retries plus one in both modes; the circuit breaker stays off unless it is enabled, and stops
the retries when it is, in both modes.
- `S3ClientProviderRetryModeTest` (new): a real SDK client against a local HTTP stub that answers
`503 SlowDown`, in both modes. A throttled request is retried until the store answers, and a request
that stays throttled fails after the configured number of retries.
- `NativeS3FileSystemFactoryTest` and `S3ClientProviderTest`: the default is `adaptive`; the option can be
set, also by the lower-case value of a Flink configuration file (`s3.retry.mode: standard`).
- All unit tests of the module pass.
- Manually, end to end, on an S3-compatible object store with a request-rate quota (details below).
**Manual verification.** A streaming SQL job, `datagen` into the `filesystem` sink (`raw` format, 1 KiB
rows, 128 MiB part files, multipart parts of about 6 MB), on Flink 2.2 on Kubernetes, checkpoint interval
5 minutes, `s3.retry.max-num-retries: 10` and the restart strategy `none`, so that a request that runs out
of attempts fails the job. The 100 GiB data was read back by a batch job that counts rows and characters
and fails on any difference. MB/s is 10^6 bytes per second. The runs below differ only in the retry mode.
Important to mention that this S3 backend is artificially configured for high throttling and low speed so
read the following numbers considering that.
100 GiB, 16 writers on 2 TaskManagers, `standard` against `adaptive`:
| | `standard` | `adaptive` | Change |
|-----------------------------------------------|------------:|------------:|-------:|
| Write duration | 17 min 15 s | 14 min 40 s | -15% |
| Write throughput | 104 MB/s | 122 MB/s | +18% |
| S3 calls (a call the SDK repeats counts once) | 20,080 | 20,096 | same |
| Throttled responses | 2,164 | 743 | -66% |
| Retried attempts | 4,351 | 1,350 | -69% |
| HTTP attempts (calls + retries) | 24,431 | 21,446 | -12% |
| Throttled share of the attempts | 8.9% | 3.5% | |
| Completed calls per second (IOPS) | 19.4 | 22.8 | +18% |
| Rejected (throttled) requests per second | 2.1 | 0.8 | -60% |
| Read-back duration | 3 min 20 s | 3 min 44 s | +12% |
| Read-back throughput | 537 MB/s | 480 MB/s | -11% |
| Read-back throttled responses | 551 | 395 | -28% |
| Read-back completed calls per second (IOPS) | 12.0 | 10.7 | -11% |
With `adaptive` the writers sent 12% fewer HTTP attempts, got 66% fewer throttled responses and finished
the write sooner, because they spent less time in backoff. The total request rate is about the same (24 HTTP
attempts per second in both runs); with `adaptive` more of them complete (19.4 to 22.8 calls per second) and
fewer are rejected (2.1 to 0.8 per second). The read-back, which has no writers to
coordinate, was not faster. These are single runs against a shared store, so the absolute rates vary from
run to run.
## Notes for reviewers
- **Internal SDK class.** The public `AdaptiveRetryStrategy.Builder` (SDK 2.44.4) has no switch for the
circuit breaker, and with the breaker left on, a throttling storm would drain its retry quota and stop
the retries even though `s3.retry.circuit-breaker.enabled` is `false`. `S3RetryStrategyFactory` therefore
casts the builder to the SDK's `DefaultAdaptiveRetryStrategy.Builder`, which has it. The circuit-breaker
tests run against both modes, so an SDK upgrade that changes this fails the build instead of the job.
- **Default.** The default changes from the standard to the adaptive strategy. If you would rather roll it
out in two steps, the default is one line in `NativeS3FileSystemFactory#RETRY_MODE`.
- **Scope of the limiter.** It belongs to one S3 client, so one TaskManager; there is no coordination
between TaskManagers, each one adapts to the throttling it sees. All requests of a client share it,
whichever bucket they go to. The AWS CRT async client has its own retries and ignores the option (said in
the option description).
- **Cost.** The limiter can delay a request before its first attempt, and a client
that is throttled on its very first requests starts from the SDK's floor of 0.5 requests per second,
which is why the stub-based tests take about 2 s longer in `adaptive` mode.
- **Not covered.** Connection errors are retried but are not throttling answers, so they do not slow the
limiter. Treating a connection
closed by the store as a throttling signal could be a follow-up.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no. The adaptive strategy is in the SDK artifact
that already provides `StandardRetryStrategy`.
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no.
`NativeS3FileSystemFactory` is `@Experimental`; it gains one option and one nested enum.
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing,
Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: yes
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? docs (the module README) and JavaDocs (the option description)
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #29499:
URL: https://github.com/apache/flink/pull/29499#issuecomment-6339219459
Just rebased to latest master + added an SDK bump to have a more stable adaptive policy. Important note that I've not yet tested it on real cluster so this is remaining. I'll have a PTO and will test it when I'm back...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "gaborgsomogyi (via GitHub) via issues" <is...@flink.apache.org>.
gaborgsomogyi commented on code in PR #29499:
URL: https://github.com/apache/flink/pull/29499#discussion_r4443813701
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderRetryModeTest.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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>"
Review Comment:
Kept `SLOW_DOWN`. It is the response of the stub, not SDK output, and the SDK has no public constant for it. Changed the assertions on SDK message text: `numAttempts()` instead of the message, and dropped the circuit breaker message check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #29499:
URL: https://github.com/apache/flink/pull/29499#issuecomment-6216804542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "171d7e7d5cd3622a11c0dc45e902dbc0d57d3823",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "171d7e7d5cd3622a11c0dc45e902dbc0d57d3823",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 171d7e7d5cd3622a11c0dc45e902dbc0d57d3823 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "gaborgsomogyi (via GitHub) via issues" <is...@flink.apache.org>.
gaborgsomogyi commented on code in PR #29499:
URL: https://github.com/apache/flink/pull/29499#discussion_r4443811549
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -312,6 +312,31 @@ 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 "
Review Comment:
Changed.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -312,6 +312,31 @@ 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
+ }
Review Comment:
Kept. Flink nests the enum of an option in the class that declares it (e.g. `JobManagerOptions.SchedulerType`), and this factory holds all the options of the module.
##########
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();
Review Comment:
No. `createAdaptive` is only called for `ADAPTIVE`, `STANDARD` uses the public `StandardRetryStrategy` builder.
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -439,6 +439,29 @@ void testRetryCircuitBreakerEnabledExplicitlyConfigured() throws Exception {
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 {
Review Comment:
Covered by `S3RetryStrategyFactoryTest`, which asserts the strategy type per mode. Added tests in `S3ClientProviderTest` that the built client uses the strategy of the mode, so this test only needs to check that the option is read.
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -439,6 +439,29 @@ void testRetryCircuitBreakerEnabledExplicitlyConfigured() throws Exception {
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 {
Review Comment:
Same as above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on PR #29499:
URL: https://github.com/apache/flink/pull/29499#issuecomment-6289021634
> > Should there be e2e test with seaweedfs for retryStrategy ?
>
> Assert on what? I've the feeling that it would add minutes to the test runtime without too much benefit. Enforcing throttling from seaweedfs is also another interesting question whether it's possible at all
Yeah ! i was having just idea if possible can there be IT. As i thought more deep into how to achieve IT. It would be overkill
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "gaborgsomogyi (via GitHub)" <gi...@apache.org>.
gaborgsomogyi commented on PR #29499:
URL: https://github.com/apache/flink/pull/29499#issuecomment-6288760757
> Should there be e2e test with seaweedfs for retryStrategy ?
Assert on what? I've the feeling that it would add minutes to the test runtime without too much benefit. Enforcing throttling from seaweedfs is also another interesting question whether it's possible at all
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "gaborgsomogyi (via GitHub) via issues" <is...@flink.apache.org>.
gaborgsomogyi commented on code in PR #29499:
URL: https://github.com/apache/flink/pull/29499#discussion_r4457429158
##########
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>
Review Comment:
I've made a conscious decision to use the latest version where there are patches and relatively stable. I've the XP that this ensure to have the most stable version which is not cutting edge with new but sometimes buggy features. You can challenge this if you think differently. My main aim is stability
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "Samrat002 (via GitHub)" <gi...@apache.org>.
Samrat002 commented on code in PR #29499:
URL: https://github.com/apache/flink/pull/29499#discussion_r4323746139
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -312,6 +312,31 @@ 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 "
Review Comment:
[NIT] Add reference to aws doc in description and this can be small to the point description.
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactory.java:
##########
@@ -312,6 +312,31 @@ 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
+ }
Review Comment:
Wondering if `NativeS3FileSystemFactory.java` right place for `RetryMode` enum ?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -439,6 +439,29 @@ void testRetryCircuitBreakerEnabledExplicitlyConfigured() throws Exception {
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 {
Review Comment:
Add assert to check if the `S3RetryStrategyFactory.create` return the intended retry class
##########
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();
Review Comment:
Will this part break if set to `STANDARD` ?
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/NativeS3FileSystemFactoryTest.java:
##########
@@ -439,6 +439,29 @@ void testRetryCircuitBreakerEnabledExplicitlyConfigured() throws Exception {
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 {
Review Comment:
Add assert to check if the S3RetryStrategyFactory.create return the intended retry class
##########
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 {
Review Comment:
👍🏻
##########
flink-filesystems/flink-s3-fs-native/src/test/java/org/apache/flink/fs/s3native/S3ClientProviderRetryModeTest.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * 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>"
Review Comment:
this string may change with sdk upgrade. Instead check if sdk has public constant or match the regex and specific word like request rate .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "Samrat002 (via GitHub) via issues" <is...@flink.apache.org>.
Samrat002 commented on code in PR #29499:
URL: https://github.com/apache/flink/pull/29499#discussion_r4457299304
##########
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>
Review Comment:
use 0.49.0 ? latest
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-41014][s3] Add an adaptive retry mode to the native S3 connector [flink]
Posted by "Samrat002 (via GitHub) via issues" <is...@flink.apache.org>.
Samrat002 merged PR #29499:
URL: https://github.com/apache/flink/pull/29499
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org