Hi Peter,
Thanks for putting together this FLIP proposal. I think this is a pragmatic way
to address this issue.
One question / observation regarding rate distribution across subtasks:
In the proposal, the global target rate is divided evenly across subtasks
(targetRate / parallelism). However, it might be that :
*
Unequal split assignment: The number of splits/partitions may not divide evenly
across subtasks (e.g., 10 partitions across 4 subtasks).
-Partition skew: Some partitions or splits naturally produce significantly
higher traffic than others.
Under an equal division strategy, subtasks handling more splits (or hot
partitions) will be capped prematurely, while subtasks with fewer or idle
splits will underutilize their quota, pulling total throughput below the target
rate.
Given that subtasks already report their individual numRecordsOutRate back to
the SourceCoordinator via ReportedThroughputEvent, have you considered:
*
weighting the allocation proportionally based on each subtask's reported
throughput (or active split count)
or
*
providing an different strategy for how the target rate is distributed across
subtasks?
Kind regards, David,
From: Peter Huang <[email protected]>
Date: Wednesday, 30 September 2026 at 19:37
To: dev <[email protected]>; Peter Huang <[email protected]>
Subject: [EXTERNAL] [Discuss] FLIP-XXX: Dynamic Rate Limiting for FLIP-27 Source
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
Unless otherwise stated above:
IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: Building C, IBM Hursley Office, Hursley Park Road,
Winchester, Hampshire SO21 2JN
