HuangZhenQiu opened a new pull request, #29627:
URL: https://github.com/apache/flink/pull/29627
## What is the purpose of the change
Adds an opt-in, dynamic rate limit for FLIP-27 sources. When checkpoints
become slow (for example under backpressure or a large backlog), the source
coordinator lowers the source rate. It raises the rate again once checkpoints
recover. This keeps checkpoint durations within a configurable fraction of the
checkpoint interval without a hand-tuned static rate limit.
The feature is disabled by default
(`source.dynamic-rate-limit.enabled=false`) and only affects sources whose
`SourceReader` is built with `RateLimiterStrategy#dynamic(double)`.
## Brief change log
- `SourceCoordinator` tracks checkpoint durations in a rolling window
(`CheckpointDurationTracker`). It collects per-subtask throughput through the
new `ReportedRecordsEvent`, and broadcasts per-subtask rates through the new
`DynamicRateLimitEvent`.
- New `RateLimitCalculator` abstraction with a default
`CheckpointDurationRateLimitCalculator`. The calculator uses the min/target/max
fractions of the checkpoint interval, or the duration trend when the interval
is unknown. It throttles down or ramps up from the observed throughput, capped
at the configured max. A custom implementation can be plugged in by class name.
- `RateAllocations` and `SubtaskRateInfo` split the global target across
subtasks, falling back to an even share when a subtask has no valid rate.
- `RateLimitedSourceReader` implements the new `SupportsDynamicRateLimit`,
which updates the per-subtask limit through the new `DynamicRateLimiter` /
`GuavaDynamicRateLimiter`.
- New `DynamicRateLimitParams` and `SourceOptions`
(`source.dynamic-rate-limit.*`). The params are wired through
`SourceTransformation`, `SourceTransformationTranslator`,
`SourceOperatorFactory`, `SourceCoordinatorProvider` and `SourceOperator`.
- `DataStreamSource#withDynamicRateLimitParams` overrides the job-wide
options for a single source.
- `CheckpointCoordinator#getCurrentCheckpointInterval` is now public so
the coordinator can read the current interval.
- The resolved settings are logged at INFO when the feature is enabled.
## Verifying this change
This change added tests and can be verified as follows:
- `CheckpointDurationRateLimitCalculatorTest`, `RateAllocationsTest`,
`DynamicRateLimitParamsTest` and `RateLimitCalculatorFactoryTest` cover the
throttle/ramp-up logic, the allocation fallbacks, and param and calculator
resolution.
- `GuavaDynamicRateLimiterTest`, `RateLimitedSourceReaderTest` and
`RateLimiterStrategyTest` cover the reader-side rate updates.
- `CheckpointDurationTrackerTest` and
`SourceCoordinatorDynamicRateLimitTest` cover coordinator behaviour: window
tracking, reported-throughput handling, and broadcasting rates to registered
readers.
- `DynamicRateLimitEventTest`, `ReportedRecordsEventTest` and
`DataStreamSourceTest` cover the events and the per-source override.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: yes (new `@PublicEvolving` `SupportsDynamicRateLimit` and
`SourceOptions`, new `RateLimiterStrategy#dynamic`, new
`DataStreamSource#withDynamicRateLimitParams`)
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes
(`RateLimitedSourceReader`, only when the dynamic strategy is used)
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
(`SourceCoordinator`; feature is off by default)
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? JavaDocs
--
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]
