Hi Weiqing, Thanks for this feedback with detailed insight.
1. Reader integration. Looks like it is an existing issue of RateLimitedSourceReader. There should be a separate PR to fix it. At the same time, the functionality needs to be added into both RateLimitedSourceReader and SourceReaderBase. 2. Recovering after backoff. It is algorithm dependent. The RateLimitCalculator is defined as an interface. FLIP-610 provides a default algorithm. A user may define a customized solution according to their needs. Please refer to the PoC for details. 3. Timeouts and restarts. The proposal mainly targets to adaptively control the flink job after it runs smoothly at the beginning. If it is not stable at the beginning, the user needs to fix the external dependencies or start with a conservative max record per second limit in the configuration. Obviously, source lag will be expected. 4. Delivering unequal allocations. The PoC defines a default balanced calculation according to the subtask throughput. Please review if it is generic enough. It is the initial PoC <https://github.com/apache/flink/pull/29627>of the proposal. Please comment on it directly. Best Regards Peter Huang On Tue, Oct 13, 2026 at 11:58 PM Weiqing Yang <[email protected]> wrote: > 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 > > > > > > > > > >
