Hi Peter, Thanks for the proposal. Using observed throughput for backoff makes sense because it avoids repeatedly reducing a target the source may never have reached. A few questions on how the pieces fit together:
1. Reader integration. The proposal adds the update hook to RateLimitedSourceReader, while SourceReaderBase already manages its own limiter. How would updates reach readers using that existing path? If wrapping is intended, how would their checkpoint callbacks and split pausing be preserved? 2. Recovering after backoff. With the activation threshold now added, I wondered about recovery after backoff. A checkpoint interval of ten minutes gives a threshold of three minutes. If throttling improves checkpoint duration from four minutes to two, would “no adjustment below the threshold” also stop the source from gradually increasing its rate again? How should that gate work alongside the relative trend check? 3. Timeouts and restarts. The feedback loop learns from completed checkpoints. What happens if a checkpoint times out before it can provide a sample? What rate would the controller start with after a coordinator restart? 4. Delivering unequal allocations. The new allocator returns a rate for each reader, but the delivery section still describes broadcasting one total and dividing it equally. How would each reader receive its individual allocation? Best, Weiqing On Tue, Oct 13, 2026 at 9:54 AM Peter Huang <[email protected]> wrote: > 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 > > > > > >
