Thanks everyone for providing valuable feedback to the FLIP draft. I will start voting early next week.
On Mon, Oct 26, 2026 at 5:21 PM Peter Huang <[email protected]> wrote: > 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 >> > > > >> > > >> > >> >
