gaborgsomogyi opened a new pull request, #29784: URL: https://github.com/apache/flink/pull/29784
## What is the purpose of the change This is a backport of https://github.com/apache/flink/pull/29499. 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 default): the behavior so far. - `adaptive`: 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 bucket and 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. The default stays `standard`, so existing deployments are not affected. A deployment that runs into a request limit can set `s3.retry.mode: adaptive`. Because the limiter stays out of the way until the store throttles, a store that does not throttle sees no difference, and where it does, requests are slowed down instead of failing. ## Brief change log - `NativeS3FileSystemFactory`: the option `s3.retry.mode` (`standard` | `adaptive`, default `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. The resolved mode is logged at INFO. - Module README: the option is added to the "Retries and Timeouts" table. - AWS SDK of the module upgraded from 2.44.4 to 2.57.2 (second commit): the limiter of the adaptive strategy was too aggressive and unstable before, which https://github.com/aws/aws-sdk-java-v2/pull/7197 fixed (first released in 2.52.0). `aws-crt` moves to 0.48.5, the version SDK 2.57.2 is built against. `apache5-client`, which every service client pulls in since SDK 2.46.0, is excluded to keep the Apache 4 based HTTP stack, because the shade configuration would not relocate its `org.apache.hc.*` classes; moving to it (`apache-client` is deprecated since SDK 2.54.15) is left for later. NOTICE is updated (`software.amazon.eventstream` became `software.amazon.awssdk:eventstream` in SDK 2.57.0). ## 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`: the default is `standard`; the option can be set, also by the lower-case value of a Flink configuration file (`s3.retry.mode: adaptive`). - `S3ClientProviderTest`: the default and the setting of the mode, and that the SDK client is built with the strategy of the configured mode and the configured number of attempts, which catches a provider that ignores the mode. - 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. They were made before the SDK upgrade of the second commit. The S3 backend is artificially configured for heavy throttling and low speed, so read the numbers with that in mind. 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.57.2) 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 stays `standard`: the AWS SDK guide recommends the standard strategy for normal use cases (https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/retry-strategy.html), so `adaptive` is opt-in. Changing the default later is one line in `NativeS3FileSystemFactory#RETRY_MODE`. - **Scope of the limiter.** The factory builds one client provider per bucket, so there is one limiter per bucket and TaskManager. That matches the AWS SDK guide, which says that the adaptive strategy assumes a client that works against a single resource, for example one S3 bucket. There is no coordination between TaskManagers, each one adapts to the throttling it sees. 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): yes. The AWS SDK of the module is upgraded from 2.44.4 to 2.57.2 and `aws-crt` from 0.45.1 to 0.48.5; `apache5-client` is excluded and NOTICE is updated. - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no. `NativeS3FileSystemFactory` has no stability annotation; 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
