[HuangZhenQiu]

What is the purpose of the change

Adds an PoC of 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

[flinkbot]

CI report:

• acb5fcb Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[HuangZhenQiu]

@flinkbot run azure