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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
