KAFKA-20828: Derive client throttling from response schema - #22908
Conversation
Generated-by: OpenAI Codex Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>
junrao
left a comment
junrao
left a comment
There was a problem hiding this comment.
@arnabnandy7 : Thanks for the PR. Left a comment.
| */ | ||
| public boolean shouldClientThrottle(short version) { | ||
| return false; | ||
| return apiKey.messageType.responseSchemas()[version].get("throttle_time_ms") != null; |
There was a problem hiding this comment.
Could we remove all the overrides that return true ?
There was a problem hiding this comment.
will work on this :)
There was a problem hiding this comment.
@junrao I've completed the removal, please check and suggest if anything else need to be remediated.
Generated-by: OpenAI Codex Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>
junrao
left a comment
junrao
left a comment
There was a problem hiding this comment.
@arnabnandy7 : Thanks for the updated PR. One more comment.
| } | ||
|
|
||
| @Test | ||
| public void testClientThrottlesResponsesWithThrottleTime() { |
There was a problem hiding this comment.
This test is ok, but it's not comprehensive. Could we test shouldClientThrottle() on all versions of all ApiKeys? shouldClientThrottle() should only be false if (1) the request version is pre KIP-219 (which we can hardcode) or (2) the response doesn't contain the throttleTimeMs field.
Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>
|
The CI failure is unrelated to this PR. |
junrao
left a comment
junrao
left a comment
There was a problem hiding this comment.
@arnabnandy7 : Thanks for the updated PR. Just a minor comment. Also, is the test failure related?
| boolean shouldClientThrottle = responseHasThrottleTime && | ||
| version >= postKip219Version.getOrDefault(apiKey, apiKey.oldestVersion()); | ||
| assertEquals(shouldClientThrottle, getResponse(apiKey, version).shouldClientThrottle(version), | ||
| apiKey + " version " + version); |
There was a problem hiding this comment.
Could we add a more descriptive error message?
There was a problem hiding this comment.
updated the message with more information as requested, @junrao
Signed-off-by: Arnab Nandy <arnab_nandy7@yahoo.com>
junrao
left a comment
junrao
left a comment
There was a problem hiding this comment.
@arnabnandy7 : Thanks for the updated PR. LGTM
|
|
||
| @Test | ||
| public void testClientThrottlesResponsesWithThrottleTime() { | ||
| Map<ApiKeys, Short> postKip219Version = Map.ofEntries( |
There was a problem hiding this comment.
Have we considered moving this map to the production code to slim down the sub-classes?
'''java
public boolean shouldClientThrottle(short version) {
Short minThrottledVersion = POST_KIP_219_VERSIONS.get(this.apiKey());
if (minThrottledVersion != null) {
return version >= minThrottledVersion;
}
return this.hasThrottleTimeMsField();
}
'''
（sorry for the poor formatting, I typed this on my phone)
There was a problem hiding this comment.
@chia7712 Thanks for the suggestion. Moving the KIP-219 version mapping into AbstractResponse would centralize the throttling logic and let us remove the remaining version-specific overrides. Since this PR has already been merged, I’ll look into addressing this in a follow-up PR.
4 participants
Description
AbstractResponse.shouldClientThrottle()previously defaulted tofalse. As a result, newer response types containing athrottleTimeMsfield did not enable client-side throttling unless they explicitly
overrode this method.
This change derives the default behavior from the response schema. A
client now throttles when the response schema for the negotiated API
version contains the
throttle_time_msfield.Existing overrides remain unchanged to preserve historical
version-specific throttling behavior for older APIs.
This fixes client-side throttling for:
ConsumerGroupHeartbeatResponseShareGroupHeartbeatResponseStreamsGroupHeartbeatResponseIt also prevents newly added response types with a throttle-time field
from accidentally omitting the required behavior.
Testing
Added regression coverage for every supported version of the affected
heartbeat APIs, verifying that
shouldClientThrottle()returnstrue.The following checks passed:
clients:test --tests org.apache.kafka.common.requests.RequestResponseTestclients:checkstyleMainclients:checkstyleTestspotlessCheckReviewers: Jun Rao junrao@gmail.com, Hardanish Singh
(github:Hardanish-Singh)