Hi Yanquan, Thanks for the feedback and questions. Sorry for the late reply as I am working on an internal solution for testing end to end.
1. Lack of source-level attribution For Flink jobs that consume data from multiple sources, users can leverage the new API introduced in DataStreamSource to configure source-specific RateLimitParams, rather than relying on a global configuration. This provides greater flexibility and enables independent rate-limiting behavior for each source based on its workload characteristics and throughput requirements. 2. Sensitivity to small absolute checkpoint durations The activation threshold is defined as a configurable fraction of the checkpoint interval. For example, if the checkpoint interval is 10 minutes and source.dynamic.rate-limit.trend-threshold is set to 0.3, the activation threshold would be 3 minutes (10 × 0.3). If the checkpoint duration remains below 3 minutes, no dynamic rate-limiting adjustment is necessary. This approach allows the activation threshold to scale proportionally with the configured checkpoint interval, avoiding unnecessary rate adjustments when checkpoint duration is within an acceptable range. 3. Coordination semantics. As you mentioned, rate updates are soft control signals and are therefore loss-tolerant. The coordination mechanism and loss-handling semantics follow the same approach as WatermarkAlignmentEvent. 4. unaligned checkpoints The proposal supports aligned checkpoints only. If users configure unaligned checkpoints, all dynamic rate-limiting configurations will be ignored and will have no effect. The Proposal is updated accordingly. Best Regards Peter Huang On Thu, Oct 8, 2026 at 3:11 AM Yanquan Lv <[email protected]> wrote: > 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 > > >
