This is an automated email from the ASF dual-hosted git repository.
junrao pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 63f445aaa96 KAFKA-20828: Derive client throttling from response schema
(#22908)
63f445aaa96 is described below
commit 63f445aaa9688f919d0606f2c6d118b6a7e3f3b3
Author: Arnab Nandy <[email protected]>
AuthorDate: Mon Jul 27 22:45:21 2026 +0530
KAFKA-20828: Derive client throttling from response schema (#22908)
`AbstractResponse.shouldClientThrottle()` previously defaulted to
`false`. As a result, newer response types containing a `throttleTimeMs`
field 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_ms` field.
Existing overrides remain unchanged to preserve historical
version-specific throttling behavior for older APIs.
This fixes client-side throttling for:
- `ConsumerGroupHeartbeatResponse`
- `ShareGroupHeartbeatResponse`
- `StreamsGroupHeartbeatResponse`
It also prevents newly added response types with a throttle-time field
from accidentally omitting the required behavior.
Reviewers: Jun Rao <[email protected]>, Hardanish Singh
(github:Hardanish-Singh)
---
.../kafka/common/requests/AbstractResponse.java | 2 +-
.../AlterPartitionReassignmentsResponse.java | 5 --
.../AlterUserScramCredentialsResponse.java | 5 --
.../common/requests/BrokerHeartbeatResponse.java | 4 --
.../requests/BrokerRegistrationResponse.java | 5 --
.../requests/DescribeTopicPartitionsResponse.java | 5 --
.../DescribeUserScramCredentialsResponse.java | 5 --
.../common/requests/ElectLeadersResponse.java | 5 --
.../ListPartitionReassignmentsResponse.java | 5 --
.../common/requests/UnregisterBrokerResponse.java | 4 --
.../kafka/common/requests/RequestResponseTest.java | 55 ++++++++++++++++++++++
11 files changed, 56 insertions(+), 44 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
index 961559d9e8f..bf265a58ca9 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java
@@ -304,7 +304,7 @@ public abstract class AbstractResponse implements
AbstractRequestResponse {
* quota violation, sends out responses before throttling.
*/
public boolean shouldClientThrottle(short version) {
- return false;
+ return
apiKey.messageType.responseSchemas()[version].get("throttle_time_ms") != null;
}
public ApiKeys apiKey() {
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsResponse.java
index 691a399761d..70063a7038e 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/AlterPartitionReassignmentsResponse.java
@@ -44,11 +44,6 @@ public class AlterPartitionReassignmentsResponse extends
AbstractResponse {
return data;
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/AlterUserScramCredentialsResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/AlterUserScramCredentialsResponse.java
index bc448a9e104..828b5939f95 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/AlterUserScramCredentialsResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/AlterUserScramCredentialsResponse.java
@@ -37,11 +37,6 @@ public class AlterUserScramCredentialsResponse extends
AbstractResponse {
return data;
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/BrokerHeartbeatResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/BrokerHeartbeatResponse.java
index f46e56ca50e..bc62d29b384 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/BrokerHeartbeatResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/BrokerHeartbeatResponse.java
@@ -59,8 +59,4 @@ public class BrokerHeartbeatResponse extends AbstractResponse
{
return new BrokerHeartbeatResponse(new
BrokerHeartbeatResponseData(readable, version));
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/BrokerRegistrationResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/BrokerRegistrationResponse.java
index be8a2f1f506..c80de014702 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/BrokerRegistrationResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/BrokerRegistrationResponse.java
@@ -59,11 +59,6 @@ public class BrokerRegistrationResponse extends
AbstractResponse {
return new BrokerRegistrationResponse(new
BrokerRegistrationResponseData(readable, version));
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public String toString() {
return data.toString();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/DescribeTopicPartitionsResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/DescribeTopicPartitionsResponse.java
index 04350480ac8..0cbc97960d3 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/DescribeTopicPartitionsResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/DescribeTopicPartitionsResponse.java
@@ -51,11 +51,6 @@ public class DescribeTopicPartitionsResponse extends
AbstractResponse {
data.setThrottleTimeMs(throttleTimeMs);
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public Map<Errors, Integer> errorCounts() {
Map<Errors, Integer> errorCounts = new EnumMap<>(Errors.class);
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/DescribeUserScramCredentialsResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/DescribeUserScramCredentialsResponse.java
index fad733fd44f..b69a77cc5e9 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/DescribeUserScramCredentialsResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/DescribeUserScramCredentialsResponse.java
@@ -37,11 +37,6 @@ public class DescribeUserScramCredentialsResponse extends
AbstractResponse {
return data;
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersResponse.java
index 1a1546980a8..46684ef9e27 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersResponse.java
@@ -83,11 +83,6 @@ public class ElectLeadersResponse extends AbstractResponse {
return new ElectLeadersResponse(new ElectLeadersResponseData(readable,
version));
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
public static Map<TopicPartition, Optional<Throwable>>
electLeadersResult(ElectLeadersResponseData data) {
Map<TopicPartition, Optional<Throwable>> map = new HashMap<>();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/ListPartitionReassignmentsResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/ListPartitionReassignmentsResponse.java
index cee49055598..bb2c645e20f 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/ListPartitionReassignmentsResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/ListPartitionReassignmentsResponse.java
@@ -42,11 +42,6 @@ public class ListPartitionReassignmentsResponse extends
AbstractResponse {
return data;
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
-
@Override
public int throttleTimeMs() {
return data.throttleTimeMs();
diff --git
a/clients/src/main/java/org/apache/kafka/common/requests/UnregisterBrokerResponse.java
b/clients/src/main/java/org/apache/kafka/common/requests/UnregisterBrokerResponse.java
index a0f71a7021f..f85f4d636d5 100644
---
a/clients/src/main/java/org/apache/kafka/common/requests/UnregisterBrokerResponse.java
+++
b/clients/src/main/java/org/apache/kafka/common/requests/UnregisterBrokerResponse.java
@@ -61,8 +61,4 @@ public class UnregisterBrokerResponse extends
AbstractResponse {
return new UnregisterBrokerResponse(new
UnregisterBrokerResponseData(readable, version));
}
- @Override
- public boolean shouldClientThrottle(short version) {
- return true;
- }
}
diff --git
a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index 3b39da0a022..a8c0d5c88db 100644
---
a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++
b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -359,6 +359,61 @@ public class RequestResponseTest {
}
}
+ @Test
+ public void testClientThrottlesResponsesWithThrottleTime() {
+ Map<ApiKeys, Short> postKip219Version = Map.ofEntries(
+ Map.entry(ApiKeys.PRODUCE, (short) 6),
+ Map.entry(ApiKeys.FETCH, (short) 8),
+ Map.entry(ApiKeys.LIST_OFFSETS, (short) 3),
+ Map.entry(ApiKeys.METADATA, (short) 6),
+ Map.entry(ApiKeys.OFFSET_COMMIT, (short) 4),
+ Map.entry(ApiKeys.OFFSET_FETCH, (short) 4),
+ Map.entry(ApiKeys.FIND_COORDINATOR, (short) 2),
+ Map.entry(ApiKeys.JOIN_GROUP, (short) 3),
+ Map.entry(ApiKeys.HEARTBEAT, (short) 2),
+ Map.entry(ApiKeys.LEAVE_GROUP, (short) 2),
+ Map.entry(ApiKeys.SYNC_GROUP, (short) 2),
+ Map.entry(ApiKeys.DESCRIBE_GROUPS, (short) 2),
+ Map.entry(ApiKeys.LIST_GROUPS, (short) 2),
+ Map.entry(ApiKeys.API_VERSIONS, (short) 2),
+ Map.entry(ApiKeys.CREATE_TOPICS, (short) 3),
+ Map.entry(ApiKeys.DELETE_TOPICS, (short) 2),
+ Map.entry(ApiKeys.DELETE_RECORDS, (short) 1),
+ Map.entry(ApiKeys.INIT_PRODUCER_ID, (short) 1),
+ Map.entry(ApiKeys.ADD_PARTITIONS_TO_TXN, (short) 1),
+ Map.entry(ApiKeys.ADD_OFFSETS_TO_TXN, (short) 1),
+ Map.entry(ApiKeys.END_TXN, (short) 1),
+ Map.entry(ApiKeys.TXN_OFFSET_COMMIT, (short) 1),
+ Map.entry(ApiKeys.DESCRIBE_ACLS, (short) 1),
+ Map.entry(ApiKeys.CREATE_ACLS, (short) 1),
+ Map.entry(ApiKeys.DELETE_ACLS, (short) 1),
+ Map.entry(ApiKeys.DESCRIBE_CONFIGS, (short) 2),
+ Map.entry(ApiKeys.ALTER_CONFIGS, (short) 1),
+ Map.entry(ApiKeys.ALTER_REPLICA_LOG_DIRS, (short) 1),
+ Map.entry(ApiKeys.DESCRIBE_LOG_DIRS, (short) 1),
+ Map.entry(ApiKeys.CREATE_PARTITIONS, (short) 1),
+ Map.entry(ApiKeys.CREATE_DELEGATION_TOKEN, (short) 1),
+ Map.entry(ApiKeys.RENEW_DELEGATION_TOKEN, (short) 1),
+ Map.entry(ApiKeys.EXPIRE_DELEGATION_TOKEN, (short) 1),
+ Map.entry(ApiKeys.DESCRIBE_DELEGATION_TOKEN, (short) 1),
+ Map.entry(ApiKeys.DELETE_GROUPS, (short) 1)
+ );
+
+ for (ApiKeys apiKey : ApiKeys.values()) {
+ for (short version : apiKey.allVersions()) {
+ boolean responseHasThrottleTime =
+
apiKey.messageType.responseSchemas()[version].get("throttle_time_ms") != null;
+ short firstPostKip219Version =
postKip219Version.getOrDefault(apiKey, apiKey.oldestVersion());
+ boolean shouldClientThrottle = responseHasThrottleTime &&
+ version >= firstPostKip219Version;
+ assertEquals(shouldClientThrottle, getResponse(apiKey,
version).shouldClientThrottle(version),
+ "Unexpected shouldClientThrottle result for " + apiKey + "
version " + version +
+ ": responseHasThrottleTime=" + responseHasThrottleTime
+
+ ", firstPostKip219Version=" + firstPostKip219Version);
+ }
+ }
+ }
+
// This test validates special cases that are not checked in
testSerialization
@Test
public void testSerializationSpecialCases() {
