Hi Peter, Thanks for putting together this proposal. I agree that static rate limiting has its limitations, especially for pipelines involving large language model (LLM) inference, where downstream processing capacity and latency can fluctuate significantly. However, I have a few questions and concerns about the current design that I would like to discuss with you and the community.
1. Lack of source-level attribution For a job with multiple sources, even if the increase in checkpoint duration is caused by only one source, every SourceCoordinator observes the duration of the same global checkpoint. As a result, all sources with dynamic rate limiting enabled may independently interpret the slowdown as a signal to reduce their rates, including sources that did not contribute to the checkpoint slowdown. 2. Sensitivity to small absolute checkpoint durations Using only the relative change in checkpoint duration can make the algorithm overly sensitive when the absolute durations are small. For example, an increase from 100 ms to 1 second represents a tenfold increase and would strongly indicate a worsening trend, even though a 1-second checkpoint may still be perfectly healthy and far below the configured checkpoint timeout. Could the algorithm introduce an activation threshold based on the checkpoint timeout, or a separately configured target duration, and ignore relative fluctuations while the checkpoint duration remains below that threshold? This could prevent unnecessary throttling caused by large relative changes from a very small baseline. 3. Coordination semantics and unaligned checkpoints The new bidirectional coordination between the SourceOperators and the SourceCoordinator also introduces additional failure-handling and compatibility concerns. The proposal does not specify how the loss of a ReportedThroughputEvent or DynamicRateLimitEvent should be handled. Since a rate update is a soft control signal, should DynamicRateLimitEvent be loss-tolerant, similar to WatermarkAlignmentEvent, rather than allowing an occasional delivery failure to trigger task failover? The proposal also does not discuss unaligned checkpoints. While the OperatorEvent mechanism itself appears to be compatible with unaligned checkpoints, the assumption behind the control signal may no longer hold. Unaligned checkpoints allow barriers to overtake buffered records and are designed to make checkpoint duration largely independent of the current throughput and backpressure. In that case, a checkpoint-duration trend may not reliably indicate that the source rate is too high. Has the algorithm been evaluated for aligned checkpoints, unaligned checkpoints, and checkpoints that switch from aligned to unaligned after an alignment timeout? Would these modes require different feedback signals or control strategies? Peter Huang <[email protected]> 于2026年10月1日周四 02:38写道： > Hi Everyone, > > I would like to start a discussion on FLIP-XXX: Dynamic Rate Limiting for > FLIP-27 Source [1]. > > Flink's FLIP-27 Source API already supports rate limiting (RateLimiter / > RateLimiterStrategy / RateLimitedSourceReader in flink-core), but the limit > is static: it's fixed once at source construction time and has no > visibility into how the job is actually running. > > Two pain points observed in this static rate limit feature. > > 1. > > It is hard to set up a static limit for different scenarios, for > example, periodical traffic peak, regional failover situation, and etc. > 2. > > For pipelines that throughput is bounded by in process RPC or downstream > service. Users prefer to slow down the process rather than impact these > services or job failure due to checkpoint timeout. > > The proposal aims to add features below to improve the resiliency of Flink > application with adjustable ratelimit with the feedback loop from > checkpoint time: > > > 1. > > Add a dynamic rate limit to the Source interface that can adjust itself > automatically, using the historical checkpoint completion time tracked > by the SourceCoordinator as the feedback signal (checkpoints getting > slower → back off; checkpoints healthy → ramp back up). > 2. > > Expose the tuning knobs as standard Flink ConfigOptions instead of a > hand-rolled parameters object, so the feature is configurable the same > way > as any other Flink job setting. > 3. > > Base the actual rate adjustment on the source operator's current > observed throughput, not on the previously-broadcast target rate, so a > throttle/ramp decision takes effect immediately instead of compounding > off > a stale number. > > > > > Looking forward to your feedback! > > > Best Regards > Peter Huang > > [1] > > https://docs.google.com/document/d/1oOUBY6cbLnbvA3Xs9Hm8irvxStPHy5jIEjFec_46rNg/edit?tab=t.0#heading=h.exegx0l01gnh >
