You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sq...@apache.org on 2026/05/06 06:35:22 UTC
(kafka) branch trunk updated: KAFKA-20551: Remove unnecessary generics from TxnOffsetCommitResponse, OffsetCommitResponse, and OffsetDeleteResponse (#22210)
This is an automated email from the ASF dual-hosted git repository.
squah-confluent 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 38227ab7682 KAFKA-20551: Remove unnecessary generics from TxnOffsetCommitResponse, OffsetCommitResponse, and OffsetDeleteResponse (#22210)
38227ab7682 is described below
commit 38227ab768282c7ed3f3986e5e2acf46219d1642
Author: Jiayao Sun <ji...@outlook.com>
AuthorDate: Wed May 6 18:35:14 2026 +1200
KAFKA-20551: Remove unnecessary generics from TxnOffsetCommitResponse, OffsetCommitResponse, and OffsetDeleteResponse (#22210)
Simplify the `addPartitions` builder methods by replacing the generic
type parameter `<P>` and the `Function` mapper with explicit request
partition classes (e.g., `OffsetCommitRequestPartition`). This removes
unnecessary abstraction that was only useful in test cases, and
streamlines the API usages across both production and test code.
Reviewers: Ken Huang <s7...@gmail.com>, Sean Quah
<sq...@confluent.io>
---
.../kafka/common/requests/OffsetCommitResponse.java | 7 ++++---
.../kafka/common/requests/OffsetDeleteResponse.java | 7 ++++---
.../common/requests/TxnOffsetCommitResponse.java | 7 ++++---
.../requests/TxnOffsetCommitResponseTest.java | 3 ++-
core/src/main/scala/kafka/server/KafkaApis.scala | 21 +++++++--------------
5 files changed, 21 insertions(+), 24 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
index 521ffa1c2fd..483640b95c4 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitResponse.java
@@ -18,6 +18,7 @@ package org.apache.kafka.common.requests;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
import org.apache.kafka.common.message.OffsetCommitResponseData;
import org.apache.kafka.common.message.OffsetCommitResponseData.OffsetCommitResponsePartition;
import org.apache.kafka.common.message.OffsetCommitResponseData.OffsetCommitResponseTopic;
@@ -165,11 +166,11 @@ public class OffsetCommitResponse extends AbstractResponse {
return this;
}
- public <P> Builder addPartitions(
+ public Builder addPartitions(
Uuid topicId,
String topicName,
- List<P> partitions,
- Function<P, Integer> partitionIndex,
+ List<OffsetCommitRequestData.OffsetCommitRequestPartition> partitions,
+ Function<OffsetCommitRequestData.OffsetCommitRequestPartition, Integer> partitionIndex,
Errors error
) {
final OffsetCommitResponseTopic topicResponse = getOrCreate(topicId, topicName);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
index 0f3655d62c6..c51b547dd58 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/OffsetDeleteResponse.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.message.OffsetDeleteRequestData;
import org.apache.kafka.common.message.OffsetDeleteResponseData;
import org.apache.kafka.common.message.OffsetDeleteResponseData.OffsetDeleteResponsePartition;
import org.apache.kafka.common.message.OffsetDeleteResponseData.OffsetDeleteResponseTopic;
@@ -75,10 +76,10 @@ public class OffsetDeleteResponse extends AbstractResponse {
return this;
}
- public <P> Builder addPartitions(
+ public Builder addPartitions(
String topicName,
- List<P> partitions,
- Function<P, Integer> partitionIndex,
+ List<OffsetDeleteRequestData.OffsetDeleteRequestPartition> partitions,
+ Function<OffsetDeleteRequestData.OffsetDeleteRequestPartition, Integer> partitionIndex,
Errors error
) {
final OffsetDeleteResponseTopic topicResponse = getOrCreateTopic(topicName);
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitResponse.java
index 5e556ad8de2..a01b5779b4e 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitResponse.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/TxnOffsetCommitResponse.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.requests;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.TxnOffsetCommitRequestData;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData.TxnOffsetCommitResponsePartition;
import org.apache.kafka.common.message.TxnOffsetCommitResponseData.TxnOffsetCommitResponseTopic;
@@ -80,10 +81,10 @@ public class TxnOffsetCommitResponse extends AbstractResponse {
return this;
}
- public <P> Builder addPartitions(
+ public Builder addPartitions(
String topicName,
- List<P> partitions,
- Function<P, Integer> partitionIndex,
+ List<TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition> partitions,
+ Function<TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition, Integer> partitionIndex,
Errors error
) {
final TxnOffsetCommitResponseTopic topicResponse = getOrCreate(topicName);
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/TxnOffsetCommitResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/TxnOffsetCommitResponseTest.java
index f3c0318de3c..6dd950a5539 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/TxnOffsetCommitResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/TxnOffsetCommitResponseTest.java
@@ -96,7 +96,8 @@ public class TxnOffsetCommitResponseTest extends OffsetCommitResponseTest {
@Test
public void testBuilderAddPartitions() {
TxnOffsetCommitResponse.Builder builder = TxnOffsetCommitResponse.newBuilder();
- builder.addPartitions(topicOne, List.of(partitionOne, partitionTwo), p -> p, errorOne);
+ builder.addPartition(topicOne, partitionOne, errorOne);
+ builder.addPartition(topicOne, partitionTwo, errorOne);
TxnOffsetCommitResponseData expected = new TxnOffsetCommitResponseData()
.setTopics(List.of(
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index d38c8921dbd..93fe6449687 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -303,13 +303,11 @@ class KafkaApis(val requestChannel: RequestChannel,
if (useTopicIds && topic.name.isEmpty) {
// If the topic name is undefined, it means that the topic id is unknown so we add
// the topic and all its partitions to the response with UNKNOWN_TOPIC_ID.
- responseBuilder.addPartitions[OffsetCommitRequestData.OffsetCommitRequestPartition](
- topic.topicId, topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_ID)
+ responseBuilder.addPartitions(topic.topicId, topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_ID)
} else if (!authorizedTopics.contains(topic.name)) {
// If the topic is not authorized, we add the topic and all its partitions
// to the response with TOPIC_AUTHORIZATION_FAILED.
- responseBuilder.addPartitions[OffsetCommitRequestData.OffsetCommitRequestPartition](
- topic.topicId, topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
+ responseBuilder.addPartitions(topic.topicId, topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
} else {
// For lower API versions, the topic id may not be included in the request.
// In this case, we resolve the topic id from metadata cache to ensure that the topic exists.
@@ -321,8 +319,7 @@ class KafkaApis(val requestChannel: RequestChannel,
if (topic.topicId == Uuid.ZERO_UUID) {
// If the topic is unknown, we add the topic and all its partitions
// to the response with UNKNOWN_TOPIC_OR_PARTITION.
- responseBuilder.addPartitions[OffsetCommitRequestData.OffsetCommitRequestPartition](
- Uuid.ZERO_UUID, topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
+ responseBuilder.addPartitions(Uuid.ZERO_UUID, topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
} else {
// Otherwise, we check all partitions to ensure that they all exist.
val topicWithValidPartitions = new OffsetCommitRequestData.OffsetCommitRequestTopic()
@@ -2063,13 +2060,11 @@ class KafkaApis(val requestChannel: RequestChannel,
if (!authorizedTopics.contains(topic.name)) {
// If the topic is not authorized, we add the topic and all its partitions
// to the response with TOPIC_AUTHORIZATION_FAILED.
- responseBuilder.addPartitions[TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition](
- topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
+ responseBuilder.addPartitions(topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
} else if (!metadataCache.contains(topic.name)) {
// If the topic is unknown, we add the topic and all its partitions
// to the response with UNKNOWN_TOPIC_OR_PARTITION.
- responseBuilder.addPartitions[TxnOffsetCommitRequestData.TxnOffsetCommitRequestPartition](
- topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
+ responseBuilder.addPartitions(topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
} else {
// Otherwise, we check all partitions to ensure that they all exist.
val topicWithValidPartitions = new TxnOffsetCommitRequestData.TxnOffsetCommitRequestTopic().setName(topic.name)
@@ -2384,13 +2379,11 @@ class KafkaApis(val requestChannel: RequestChannel,
if (!authorizedTopics.contains(topic.name)) {
// If the topic is not authorized, we add the topic and all its partitions
// to the response with TOPIC_AUTHORIZATION_FAILED.
- responseBuilder.addPartitions[OffsetDeleteRequestData.OffsetDeleteRequestPartition](
- topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
+ responseBuilder.addPartitions(topic.name, topic.partitions, _.partitionIndex, Errors.TOPIC_AUTHORIZATION_FAILED)
} else if (!metadataCache.contains(topic.name)) {
// If the topic is unknown, we add the topic and all its partitions
// to the response with UNKNOWN_TOPIC_OR_PARTITION.
- responseBuilder.addPartitions[OffsetDeleteRequestData.OffsetDeleteRequestPartition](
- topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
+ responseBuilder.addPartitions(topic.name, topic.partitions, _.partitionIndex, Errors.UNKNOWN_TOPIC_OR_PARTITION)
} else {
// Otherwise, we check all partitions to ensure that they all exist.
val topicWithValidPartitions = new OffsetDeleteRequestData.OffsetDeleteRequestTopic().setName(topic.name)