You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2023/10/19 14:48:19 UTC
[kafka] branch trunk updated: KAFKA-15454: Add support for OffsetCommit version 9 in admin client (#14571)
This is an automated email from the ASF dual-hosted git repository.
dajac 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 8f3731e2bd1 KAFKA-15454: Add support for OffsetCommit version 9 in admin client (#14571)
8f3731e2bd1 is described below
commit 8f3731e2bd14485ab6a5e29afe9bfbfdc1aee432
Author: vamossagar12 <sa...@gmail.com>
AuthorDate: Thu Oct 19 20:18:12 2023 +0530
KAFKA-15454: Add support for OffsetCommit version 9 in admin client (#14571)
This patch adds support for OffsetCommit version 9 in the admin client. It mainly allows handling two new error codes `STALE_MEMBER_EPOCH` and `GROUP_ID_NOT_FOUND ` introduced as part of KIP-848.
Reviewers: David Jacot <dj...@confluent.io>
---
.../kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java | 2 ++
.../java/org/apache/kafka/common/requests/OffsetCommitResponse.java | 2 ++
.../test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java | 2 +-
3 files changed, 5 insertions(+), 1 deletion(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
index 425ed66bd29..d918fd28aa1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
@@ -179,8 +179,10 @@ public class AlterConsumerGroupOffsetsHandler extends AdminApiHandler.Batched<Co
case INVALID_GROUP_ID:
case INVALID_COMMIT_OFFSET_SIZE:
case GROUP_AUTHORIZATION_FAILED:
+ case GROUP_ID_NOT_FOUND:
// Member level errors.
case UNKNOWN_MEMBER_ID:
+ case STALE_MEMBER_EPOCH:
log.debug("OffsetCommit request for group id {} failed due to error {}.",
groupId.idValue, error);
partitionResults.put(topicPartition, error);
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 a87e506fdd4..9848ad380fa 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
@@ -48,6 +48,8 @@ import java.util.function.Function;
* - {@link Errors#GROUP_AUTHORIZATION_FAILED}
* - {@link Errors#INVALID_PRODUCER_ID_MAPPING}
* - {@link Errors#INVALID_TXN_STATE}
+ * - {@link Errors#GROUP_ID_NOT_FOUND}
+ * - {@link Errors#STALE_MEMBER_EPOCH}
*/
public class OffsetCommitResponse extends AbstractResponse {
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 378e08b2c45..b1caf607ad4 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -4757,7 +4757,7 @@ public class KafkaAdminClientTest {
final TopicPartition tp1 = new TopicPartition("foo", 0);
final List<Errors> nonRetriableErrors = Arrays.asList(
- Errors.GROUP_AUTHORIZATION_FAILED, Errors.INVALID_GROUP_ID, Errors.GROUP_ID_NOT_FOUND);
+ Errors.GROUP_AUTHORIZATION_FAILED, Errors.INVALID_GROUP_ID, Errors.GROUP_ID_NOT_FOUND, Errors.STALE_MEMBER_EPOCH);
try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(mockCluster(1, 0))) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());