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/12/04 15:19:37 UTC
(kafka) branch trunk updated: KAFKA-14509: [2/N] Implement server side logic for ConsumerGroupDescribe API (#14544)
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 b7c99e22a77 KAFKA-14509: [2/N] Implement server side logic for ConsumerGroupDescribe API (#14544)
b7c99e22a77 is described below
commit b7c99e22a77392d6053fe231209e1de32b50a98b
Author: Max Riedel <29...@users.noreply.github.com>
AuthorDate: Mon Dec 4 16:19:28 2023 +0100
KAFKA-14509: [2/N] Implement server side logic for ConsumerGroupDescribe API (#14544)
This patch implements the ConsumerGroupDescribe API.
Reviewers: David Jacot <dj...@confluent.io>
---
.../requests/ConsumerGroupDescribeRequest.java | 13 ++
.../message/ConsumerGroupDescribeResponse.json | 2 +-
.../requests/ConsumerGroupDescribeRequestTest.java | 23 ++++
.../group/GroupCoordinatorAdapter.scala | 11 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 44 ++++++-
.../group/GroupCoordinatorAdapterTest.scala | 13 ++
.../scala/unit/kafka/server/KafkaApisTest.scala | 80 ++++++++++++
.../kafka/coordinator/group/GroupCoordinator.java | 18 ++-
.../coordinator/group/GroupCoordinatorService.java | 63 ++++++++++
.../coordinator/group/GroupCoordinatorShard.java | 16 +++
.../coordinator/group/GroupMetadataManager.java | 59 ++++++++-
.../coordinator/group/consumer/ConsumerGroup.java | 25 +++-
.../group/consumer/ConsumerGroupMember.java | 35 ++++++
.../group/GroupCoordinatorServiceTest.java | 138 +++++++++++++++++++++
.../group/GroupMetadataManagerTest.java | 117 +++++++++++++++++
.../group/consumer/ConsumerGroupMemberTest.java | 84 +++++++++++++
.../group/consumer/ConsumerGroupTest.java | 35 ++++++
17 files changed, 766 insertions(+), 10 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequest.java
index 862d9c9d4e4..48a03bd8bb6 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequest.java
@@ -23,6 +23,8 @@ import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.Errors;
import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.stream.Collectors;
public class ConsumerGroupDescribeRequest extends AbstractRequest {
@@ -83,4 +85,15 @@ public class ConsumerGroupDescribeRequest extends AbstractRequest {
version
);
}
+
+ public static List<ConsumerGroupDescribeResponseData.DescribedGroup> getErrorDescribedGroupList(
+ List<String> groupIds,
+ Errors error
+ ) {
+ return groupIds.stream()
+ .map(groupId -> new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setErrorCode(error.code())
+ ).collect(Collectors.toList());
+ }
}
diff --git a/clients/src/main/resources/common/message/ConsumerGroupDescribeResponse.json b/clients/src/main/resources/common/message/ConsumerGroupDescribeResponse.json
index b0c98b78c1a..f6839b2250c 100644
--- a/clients/src/main/resources/common/message/ConsumerGroupDescribeResponse.json
+++ b/clients/src/main/resources/common/message/ConsumerGroupDescribeResponse.json
@@ -50,7 +50,7 @@
{ "name": "Members", "type": "[]Member", "versions": "0+",
"about": "The members.",
"fields": [
- { "name": "MemberId", "type": "uuid", "versions": "0+",
+ { "name": "MemberId", "type": "string", "versions": "0+",
"about": "The member ID." },
{ "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
"about": "The member instance ID." },
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequestTest.java
index 81255da2b11..23ea03c1730 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/ConsumerGroupDescribeRequestTest.java
@@ -24,6 +24,7 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
+import static org.apache.kafka.common.requests.ConsumerGroupDescribeRequest.getErrorDescribedGroupList;
import static org.junit.jupiter.api.Assertions.assertEquals;
public class ConsumerGroupDescribeRequestTest {
@@ -47,4 +48,26 @@ public class ConsumerGroupDescribeRequestTest {
assertEquals(Errors.forException(e).code(), group.errorCode());
}
}
+
+ @Test
+ public void testGetErrorDescribedGroupList() {
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expectedDescribedGroupList = Arrays.asList(
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-1")
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code()),
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-2")
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code()),
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-3")
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code())
+ );
+
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> describedGroupList = getErrorDescribedGroupList(
+ Arrays.asList("group-id-1", "group-id-2", "group-id-3"),
+ Errors.COORDINATOR_LOAD_IN_PROGRESS
+ );
+
+ assertEquals(expectedDescribedGroupList, describedGroupList);
+ }
}
diff --git a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
index 409c1e1d849..678844cc662 100644
--- a/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
+++ b/core/src/main/scala/kafka/coordinator/group/GroupCoordinatorAdapter.scala
@@ -20,7 +20,7 @@ import kafka.common.OffsetAndMetadata
import kafka.server.{KafkaConfig, ReplicaManager, RequestLocal}
import kafka.utils.Implicits.MapExtensionMethods
import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
-import org.apache.kafka.common.message.{ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchRequestData, OffsetFetchRespon [...]
+import org.apache.kafka.common.message.{ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsResponseData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupRequestData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, Offset [...]
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.RecordBatch
@@ -601,4 +601,13 @@ private[group] class GroupCoordinatorAdapter(
override def shutdown(): Unit = {
coordinator.shutdown()
}
+
+ override def consumerGroupDescribe(
+ context: RequestContext,
+ groupIds: util.List[String]
+ ): CompletableFuture[util.List[ConsumerGroupDescribeResponseData.DescribedGroup]] = {
+ FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
+ s"The old group coordinator does not support ${ApiKeys.CONSUMER_GROUP_DESCRIBE.name} API."
+ ))
+ }
}
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index cb912769f17..1ac38a0895d 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -3731,8 +3731,48 @@ class KafkaApis(val requestChannel: RequestChannel,
}
def handleConsumerGroupDescribe(request: RequestChannel.Request): CompletableFuture[Unit] = {
- requestHelper.sendMaybeThrottle(request, request.body[ConsumerGroupDescribeRequest].getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
- CompletableFuture.completedFuture[Unit](())
+ val consumerGroupDescribeRequest = request.body[ConsumerGroupDescribeRequest]
+
+ if (!config.isNewGroupCoordinatorEnabled) {
+ // The API is not supported by the "old" group coordinator (the default). If the
+ // new one is not enabled, we fail directly here.
+ requestHelper.sendMaybeThrottle(request, request.body[ConsumerGroupDescribeRequest].getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else {
+ val response = new ConsumerGroupDescribeResponseData()
+
+ val authorizedGroups = new ArrayBuffer[String]()
+ consumerGroupDescribeRequest.data.groupIds.forEach { groupId =>
+ if (!authHelper.authorize(request.context, DESCRIBE, GROUP, groupId)) {
+ response.groups.add(new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setErrorCode(Errors.GROUP_AUTHORIZATION_FAILED.code)
+ )
+ } else {
+ authorizedGroups += groupId
+ }
+ }
+
+ groupCoordinator.consumerGroupDescribe(
+ request.context,
+ authorizedGroups.asJava
+ ).handle[Unit] { (results, exception) =>
+ if (exception != null) {
+ requestHelper.sendMaybeThrottle(request, consumerGroupDescribeRequest.getErrorResponse(exception))
+ } else {
+ if (response.groups.isEmpty) {
+ // If the response is empty, we can directly reuse the results.
+ response.setGroups(results)
+ } else {
+ // Otherwise, we have to copy the results into the existing ones.
+ response.groups.addAll(results)
+ }
+
+ requestHelper.sendMaybeThrottle(request, new ConsumerGroupDescribeResponse(response))
+ }
+ }
+ }
+
}
// Just a place holder for now.
diff --git a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
index 75ecb1e101b..022a8716d83 100644
--- a/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
+++ b/core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorAdapterTest.scala
@@ -881,4 +881,17 @@ class GroupCoordinatorAdapterTest {
assertTrue(future.isCompletedExceptionally)
assertFutureThrows(future, classOf[InvalidGroupIdException])
}
+
+ @Test
+ def testConsumerGroupDescribe(): Unit = {
+ val groupCoordinator = mock(classOf[GroupCoordinator])
+ val adapter = new GroupCoordinatorAdapter(groupCoordinator, Time.SYSTEM)
+ val context = makeContext(ApiKeys.CONSUMER_GROUP_DESCRIBE, ApiKeys.CONSUMER_GROUP_DESCRIBE.latestVersion)
+ val groupIds = List("group-id-1", "group-id-2").asJava
+
+ val future = adapter.consumerGroupDescribe(context, groupIds)
+ assertTrue(future.isDone)
+ assertTrue(future.isCompletedExceptionally)
+ assertFutureThrows(future, classOf[UnsupportedVersionException])
+ }
}
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index db200da8e33..82ef89d10b7 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -6731,12 +6731,45 @@ class KafkaApisTest {
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.errorCode)
}
+ @Test
+ def testConsumerGroupDescribe(): Unit = {
+ val groupIds = List("group-id-0", "group-id-1", "group-id-2").asJava
+ val consumerGroupDescribeRequestData = new ConsumerGroupDescribeRequestData()
+ consumerGroupDescribeRequestData.groupIds.addAll(groupIds)
+ val requestChannelRequest = buildRequest(new ConsumerGroupDescribeRequest.Builder(consumerGroupDescribeRequestData, true).build())
+
+ val future = new CompletableFuture[util.List[ConsumerGroupDescribeResponseData.DescribedGroup]]()
+ when(groupCoordinator.consumerGroupDescribe(
+ any[RequestContext],
+ any[util.List[String]]
+ )).thenReturn(future)
+
+ createKafkaApis(
+ overrideProperties = Map(KafkaConfig.NewGroupCoordinatorEnableProp -> "true")
+ ).handle(requestChannelRequest, RequestLocal.NoCaching)
+
+ val describedGroups = List(
+ new DescribedGroup().setGroupId(groupIds.get(0)),
+ new DescribedGroup().setGroupId(groupIds.get(1)),
+ new DescribedGroup().setGroupId(groupIds.get(2))
+ ).asJava
+
+ future.complete(describedGroups)
+ val expectedConsumerGroupDescribeResponseData = new ConsumerGroupDescribeResponseData()
+ .setGroups(describedGroups)
+
+ val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
+
+ assertEquals(expectedConsumerGroupDescribeResponseData, response.data)
+ }
+
@Test
def testConsumerGroupDescribeReturnsUnsupportedVersion(): Unit = {
val groupId = "group0"
val consumerGroupDescribeRequestData = new ConsumerGroupDescribeRequestData()
consumerGroupDescribeRequestData.groupIds.add(groupId)
val requestChannelRequest = buildRequest(new ConsumerGroupDescribeRequest.Builder(consumerGroupDescribeRequestData, true).build())
+
val errorCode = Errors.UNSUPPORTED_VERSION.code
val expectedDescribedGroup = new DescribedGroup().setGroupId(groupId).setErrorCode(errorCode)
val expectedResponse = new ConsumerGroupDescribeResponseData()
@@ -6748,6 +6781,53 @@ class KafkaApisTest {
assertEquals(expectedResponse, response.data)
}
+ @Test
+ def testConsumerGroupDescribeAuthorizationFailed(): Unit = {
+ val consumerGroupDescribeRequestData = new ConsumerGroupDescribeRequestData()
+ consumerGroupDescribeRequestData.groupIds.add("group-id")
+ val requestChannelRequest = buildRequest(new ConsumerGroupDescribeRequest.Builder(consumerGroupDescribeRequestData, true).build())
+
+ val authorizer: Authorizer = mock(classOf[Authorizer])
+ when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
+ .thenReturn(Seq(AuthorizationResult.DENIED).asJava)
+
+ val future = new CompletableFuture[util.List[ConsumerGroupDescribeResponseData.DescribedGroup]]()
+ when(groupCoordinator.consumerGroupDescribe(
+ any[RequestContext],
+ any[util.List[String]]
+ )).thenReturn(future)
+ future.complete(List().asJava)
+
+ createKafkaApis(
+ authorizer = Some(authorizer),
+ overrideProperties = Map(KafkaConfig.NewGroupCoordinatorEnableProp -> "true")
+ ).handle(requestChannelRequest, RequestLocal.NoCaching)
+
+ val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
+ assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code, response.data.groups.get(0).errorCode)
+ }
+
+ @Test
+ def testConsumerGroupDescribeFutureFailed(): Unit = {
+ val consumerGroupDescribeRequestData = new ConsumerGroupDescribeRequestData()
+ consumerGroupDescribeRequestData.groupIds.add("group-id")
+ val requestChannelRequest = buildRequest(new ConsumerGroupDescribeRequest.Builder(consumerGroupDescribeRequestData, true).build())
+
+ val future = new CompletableFuture[util.List[ConsumerGroupDescribeResponseData.DescribedGroup]]()
+ when(groupCoordinator.consumerGroupDescribe(
+ any[RequestContext],
+ any[util.List[String]]
+ )).thenReturn(future)
+
+ createKafkaApis(overrideProperties = Map(
+ KafkaConfig.NewGroupCoordinatorEnableProp -> "true"
+ )).handle(requestChannelRequest, RequestLocal.NoCaching)
+
+ future.completeExceptionally(Errors.FENCED_MEMBER_EPOCH.exception)
+ val response = verifyNoThrottling[ConsumerGroupDescribeResponse](requestChannelRequest)
+ assertEquals(Errors.FENCED_MEMBER_EPOCH.code, response.data.groups.get(0).errorCode)
+ }
+
@Test
def testGetTelemetrySubscriptionsNotAllowedForZkClusters(): Unit = {
val data = new GetTelemetrySubscriptionsRequestData()
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
index 291afdc5bd3..948a58516c0 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinator.java
@@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.DeleteGroupsResponseData;
@@ -25,12 +26,12 @@ import org.apache.kafka.common.message.HeartbeatRequestData;
import org.apache.kafka.common.message.HeartbeatResponseData;
import org.apache.kafka.common.message.JoinGroupRequestData;
import org.apache.kafka.common.message.JoinGroupResponseData;
-import org.apache.kafka.common.message.OffsetCommitRequestData;
-import org.apache.kafka.common.message.OffsetCommitResponseData;
import org.apache.kafka.common.message.LeaveGroupRequestData;
import org.apache.kafka.common.message.LeaveGroupResponseData;
import org.apache.kafka.common.message.ListGroupsRequestData;
import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.common.message.OffsetCommitRequestData;
+import org.apache.kafka.common.message.OffsetCommitResponseData;
import org.apache.kafka.common.message.OffsetDeleteRequestData;
import org.apache.kafka.common.message.OffsetDeleteResponseData;
import org.apache.kafka.common.message.OffsetFetchRequestData;
@@ -158,6 +159,19 @@ public interface GroupCoordinator {
List<String> groupIds
);
+ /**
+ * Describe consumer groups.
+ *
+ * @param context The coordinator request context.
+ * @param groupIds The group ids.
+ *
+ * @return A future yielding the results or an exception.
+ */
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consumerGroupDescribe(
+ RequestContext context,
+ List<String> groupIds
+ );
+
/**
* Delete Groups.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
index 817f063b6ae..a05402d410b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
@@ -30,6 +30,7 @@ import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.internals.Topic;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.DeleteGroupsResponseData;
@@ -55,6 +56,7 @@ import org.apache.kafka.common.message.TxnOffsetCommitResponseData;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.DescribeGroupsRequest;
import org.apache.kafka.common.requests.DeleteGroupsRequest;
+import org.apache.kafka.common.requests.ConsumerGroupDescribeRequest;
import org.apache.kafka.common.requests.OffsetCommitRequest;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.common.requests.TransactionResult;
@@ -547,6 +549,67 @@ public class GroupCoordinatorService implements GroupCoordinator {
return future;
}
+ /**
+ * See {@link GroupCoordinator#consumerGroupDescribe(RequestContext, List)}.
+ */
+ @Override
+ public CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> consumerGroupDescribe(
+ RequestContext context,
+ List<String> groupIds
+ ) {
+ if (!isActive.get()) {
+ return CompletableFuture.completedFuture(ConsumerGroupDescribeRequest.getErrorDescribedGroupList(
+ groupIds,
+ Errors.COORDINATOR_NOT_AVAILABLE
+ ));
+ }
+
+ final List<CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>>> futures =
+ new ArrayList<>(groupIds.size());
+ final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
+ groupIds.forEach(groupId -> {
+ if (isGroupIdNotEmpty(groupId)) {
+ groupsByTopicPartition
+ .computeIfAbsent(topicPartitionFor(groupId), __ -> new ArrayList<>())
+ .add(groupId);
+ } else {
+ futures.add(CompletableFuture.completedFuture(Collections.singletonList(
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code())
+ )));
+ }
+ });
+
+ groupsByTopicPartition.forEach((topicPartition, groupList) -> {
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
+ runtime.scheduleReadOperation(
+ "consumer-group-describe",
+ topicPartition,
+ (coordinator, lastCommittedOffset) -> coordinator.consumerGroupDescribe(groupIds, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ConsumerGroupDescribe request {} hit an unexpected exception: {}.",
+ groupList, exception.getMessage());
+ }
+
+ return ConsumerGroupDescribeRequest.getErrorDescribedGroupList(
+ groupList,
+ Errors.forException(exception)
+ );
+ });
+
+ futures.add(future);
+ });
+
+ final CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
+ return allFutures.thenApply(v -> {
+ final List<ConsumerGroupDescribeResponseData.DescribedGroup> res = new ArrayList<>();
+ futures.forEach(future -> res.addAll(future.join()));
+ return res;
+ });
+ }
+
/**
* See {@link GroupCoordinator#describeGroups(RequestContext, List)}.
*/
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
index 40b3b317536..5c5af61634c 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.coordinator.group;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -460,6 +461,21 @@ public class GroupCoordinatorShard implements CoordinatorShard<Record> {
return groupMetadataManager.listGroups(statesFilter, committedOffset);
}
+ /**
+ * Handles a ConsumerGroupDescribe request.
+ *
+ * @param groupIds The IDs of the groups to describe.
+ *
+ * @return A list containing the ConsumerGroupDescribeResponseData.DescribedGroup.
+ *
+ */
+ public List<ConsumerGroupDescribeResponseData.DescribedGroup> consumerGroupDescribe(
+ List<String> groupIds,
+ long committedOffset
+ ) {
+ return groupMetadataManager.consumerGroupDescribe(groupIds, committedOffset);
+ }
+
/**
* Handles a DescribeGroups request.
*
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index eafb6bb2170..1de63d714f9 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -30,6 +30,7 @@ import org.apache.kafka.common.errors.UnknownServerException;
import org.apache.kafka.common.errors.UnsupportedAssignorException;
import org.apache.kafka.common.errors.FencedInstanceIdException;
import org.apache.kafka.common.errors.UnreleasedInstanceIdException;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.DescribeGroupsResponseData;
@@ -270,7 +271,6 @@ public class GroupMetadataManager {
);
}
}
-
/**
* The log context.
*/
@@ -468,6 +468,34 @@ public class GroupMetadataManager {
return groupStream.map(group -> group.asListedGroup(committedOffset)).collect(Collectors.toList());
}
+
+ /**
+ * Handles a ConsumerGroupDescribe request.
+ *
+ * @param groupIds The IDs of the groups to describe.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A list containing the ConsumerGroupDescribeResponseData.DescribedGroup.
+ */
+ public List<ConsumerGroupDescribeResponseData.DescribedGroup> consumerGroupDescribe(
+ List<String> groupIds,
+ long committedOffset
+ ) {
+ final List<ConsumerGroupDescribeResponseData.DescribedGroup> describedGroups = new ArrayList<>();
+ groupIds.forEach(groupId -> {
+ try {
+ describedGroups.add(consumerGroup(groupId, committedOffset).asDescribedGroup(committedOffset, defaultAssignor.name()));
+ } catch (GroupIdNotFoundException exception) {
+ describedGroups.add(new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code())
+ );
+ }
+ });
+
+ return describedGroups;
+ }
+
/**
* Handles a DescribeGroup request.
*
@@ -623,6 +651,31 @@ public class GroupMetadataManager {
}
}
+ /**
+ * Gets a consumer group by committed offset.
+ *
+ * @param groupId The group id.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A ConsumerGroup.
+ * @throws GroupIdNotFoundException if the group does not exist or is not a consumer group.
+ */
+ public ConsumerGroup consumerGroup(
+ String groupId,
+ long committedOffset
+ ) throws GroupIdNotFoundException {
+ Group group = group(groupId, committedOffset);
+
+ if (group.type() == CONSUMER) {
+ return (ConsumerGroup) group;
+ } else {
+ // We don't support upgrading/downgrading between protocols at the moment so
+ // we throw an exception if a group exists with the wrong type.
+ throw new GroupIdNotFoundException(String.format("Group %s is not a consumer group.",
+ groupId));
+ }
+ }
+
/**
* Removes the group.
*
@@ -2505,7 +2558,7 @@ public class GroupMetadataManager {
if (group.isInState(PREPARING_REBALANCE) && group.previousState() == EMPTY) {
group.setNewMemberAdded(true);
}
-
+
group.add(member, responseFuture);
// The session timeout does not affect new members since they do not have their memberId and
@@ -2648,7 +2701,7 @@ public class GroupMetadataManager {
/**
* Reset assignment for all members and propagate the error to all members in the group.
- *
+ *
* @param group The group.
* @param error The error to propagate.
*/
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
index bad716d82ff..0fe5555ac31 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ListGroupsResponseData;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.coordinator.group.Group;
@@ -520,7 +521,14 @@ public class ConsumerGroup implements Group {
* @return The preferred assignor for the group.
*/
public Optional<String> preferredServerAssignor() {
- return serverAssignors.entrySet().stream()
+ return preferredServerAssignor(Long.MAX_VALUE);
+ }
+
+ /**
+ * @return The preferred assignor for the group with given offset.
+ */
+ public Optional<String> preferredServerAssignor(long committedOffset) {
+ return serverAssignors.entrySet(committedOffset).stream()
.max(Map.Entry.comparingByValue())
.map(Map.Entry::getKey);
}
@@ -943,4 +951,19 @@ public class ConsumerGroup implements Group {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ public ConsumerGroupDescribeResponseData.DescribedGroup asDescribedGroup(long committedOffset, String defaultAssignor) {
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setAssignorName(preferredServerAssignor(committedOffset).orElse(defaultAssignor))
+ .setGroupEpoch(groupEpoch.get(committedOffset))
+ .setGroupState(state.get(committedOffset).toString())
+ .setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset));
+ members.entrySet(committedOffset).forEach(
+ entry -> describedGroup.members().add(
+ entry.getValue().asConsumerGroupDescribeMember(targetAssignment.get(entry.getValue().memberId(), committedOffset))
+ )
+ );
+ return describedGroup;
+ }
}
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
index 729edf8d06b..9110a24c627 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMember.java
@@ -17,9 +17,11 @@
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
@@ -545,6 +547,39 @@ public class ConsumerGroupMember {
')';
}
+ /**
+ * @param targetAssignment The target assignment of this member in the corresponding group.
+ *
+ * @return The ConsumerGroupMember mapped as ConsumerGroupDescribeResponseData.Member.
+ */
+ public ConsumerGroupDescribeResponseData.Member asConsumerGroupDescribeMember(Assignment targetAssignment) {
+ return new ConsumerGroupDescribeResponseData.Member()
+ .setMemberEpoch(memberEpoch)
+ .setMemberId(memberId)
+ .setAssignment(new ConsumerGroupDescribeResponseData.Assignment()
+ .setTopicPartitions(topicPartitionsFromMap(assignedPartitions)))
+ .setTargetAssignment(new ConsumerGroupDescribeResponseData.Assignment()
+ .setTopicPartitions(topicPartitionsFromMap(
+ targetAssignment != null ? targetAssignment.partitions() : Collections.emptyMap()
+ )))
+ .setClientHost(clientHost)
+ .setClientId(clientId)
+ .setInstanceId(instanceId)
+ .setRackId(rackId)
+ .setSubscribedTopicNames(subscribedTopicNames)
+ .setSubscribedTopicRegex(subscribedTopicRegex);
+ }
+
+ private static List<ConsumerGroupDescribeResponseData.TopicPartitions> topicPartitionsFromMap(
+ Map<Uuid, Set<Integer>> partitions
+ ) {
+ return partitions.entrySet().stream().map(
+ item -> new ConsumerGroupDescribeResponseData.TopicPartitions()
+ .setTopicId(item.getKey())
+ .setPartitions(new ArrayList<>(item.getValue()))
+ ).collect(Collectors.toList());
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) return true;
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
index 752cca39f3e..4e0e2b2d573 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java
@@ -32,6 +32,7 @@ import org.apache.kafka.common.errors.RecordBatchTooLargeException;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
import org.apache.kafka.common.message.DescribeGroupsResponseData;
@@ -1316,6 +1317,143 @@ public class GroupCoordinatorServiceTest {
);
}
+ @Test
+ public void testConsumerGroupDescribe() throws InterruptedException, ExecutionException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ int partitionCount = 2;
+ service.startup(() -> partitionCount);
+
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup1 = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-1");
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup2 = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-2");
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
+ describedGroup1,
+ describedGroup2
+ );
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("consumer-group-describe"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(describedGroup1)));
+
+ CompletableFuture<Object> describedGroupFuture = new CompletableFuture<>();
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("consumer-group-describe"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 1)),
+ ArgumentMatchers.any()
+ )).thenReturn(describedGroupFuture);
+
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
+ service.consumerGroupDescribe(requestContext(ApiKeys.CONSUMER_GROUP_DESCRIBE), Arrays.asList("group-id-1", "group-id-2"));
+
+ assertFalse(future.isDone());
+ describedGroupFuture.complete(Collections.singletonList(describedGroup2));
+ assertEquals(expectedDescribedGroups, future.get());
+ }
+
+ @Test
+ public void testConsumerGroupDescribeInvalidGroupId() throws ExecutionException, InterruptedException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ int partitionCount = 1;
+ service.startup(() -> partitionCount);
+
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code());
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code()),
+ describedGroup
+ );
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("consumer-group-describe"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(describedGroup)));
+
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
+ service.consumerGroupDescribe(requestContext(ApiKeys.CONSUMER_GROUP_DESCRIBE), Arrays.asList("", null));
+
+ assertEquals(expectedDescribedGroups, future.get());
+ }
+
+ @Test
+ public void testConsumerGroupDescribeCoordinatorLoadInProgress() throws ExecutionException, InterruptedException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ int partitionCount = 1;
+ service.startup(() -> partitionCount);
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("consumer-group-describe"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(
+ new CoordinatorLoadInProgressException(null)
+ ));
+
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
+ service.consumerGroupDescribe(requestContext(ApiKeys.CONSUMER_GROUP_DESCRIBE), Collections.singletonList("group-id"));
+
+ assertEquals(
+ Collections.singletonList(new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id")
+ .setErrorCode(Errors.COORDINATOR_LOAD_IN_PROGRESS.code())
+ ),
+ future.get()
+ );
+ }
+
+ @Test
+ public void testConsumerGroupDescribeCoordinatorNotActive() throws ExecutionException, InterruptedException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime,
+ new GroupCoordinatorMetrics()
+ );
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("consumer-group-describe"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(
+ Errors.COORDINATOR_NOT_AVAILABLE.exception()
+ ));
+
+ CompletableFuture<List<ConsumerGroupDescribeResponseData.DescribedGroup>> future =
+ service.consumerGroupDescribe(requestContext(ApiKeys.CONSUMER_GROUP_DESCRIBE), Collections.singletonList("group-id"));
+
+ assertEquals(
+ Collections.singletonList(new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id")
+ .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
+ ),
+ future.get()
+ );
+ }
+
@Test
public void testDeleteOffsets() throws Exception {
CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 6711619eb09..1bfbf4ff8ac 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -31,6 +31,7 @@ import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.errors.UnknownServerException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.errors.UnsupportedAssignorException;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.errors.UnreleasedInstanceIdException;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatRequestData;
import org.apache.kafka.common.message.ConsumerGroupHeartbeatResponseData;
@@ -1044,6 +1045,10 @@ public class GroupMetadataManagerTest {
return groupMetadataManager.listGroups(statesFilter, lastCommittedOffset);
}
+ public List<ConsumerGroupDescribeResponseData.DescribedGroup> sendConsumerGroupDescribe(List<String> groupIds) {
+ return groupMetadataManager.consumerGroupDescribe(groupIds, lastCommittedOffset);
+ }
+
public List<DescribeGroupsResponseData.DescribedGroup> describeGroups(List<String> groupIds) {
return groupMetadataManager.describeGroups(groupIds, lastCommittedOffset);
}
@@ -9603,6 +9608,118 @@ public class GroupMetadataManagerTest {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testConsumerGroupDescribeNoErrors() {
+ List<String> consumerGroupIds = Arrays.asList("group-id-1", "group-id-2");
+ int epoch = 10;
+ String memberId = "member-id";
+ String topicName = "topicName";
+ ConsumerGroupMember.Builder memberBuilder = new ConsumerGroupMember.Builder(memberId)
+ .setSubscribedTopicNames(Collections.singletonList(topicName))
+ .setServerAssignorName("assignorName");
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupIds.get(0), epoch))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupIds.get(1), epoch)
+ .withMember(memberBuilder.build()))
+ .build();
+
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Arrays.asList(
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupEpoch(epoch)
+ .setGroupId(consumerGroupIds.get(0))
+ .setGroupState("empty")
+ .setAssignorName("range"),
+ new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupEpoch(epoch)
+ .setGroupId(consumerGroupIds.get(1))
+ .setMembers(Arrays.asList(
+ memberBuilder.build().asConsumerGroupDescribeMember(new Assignment(Collections.emptyMap()))
+ ))
+ .setGroupState("assigning")
+ .setAssignorName("assignorName")
+ );
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(consumerGroupIds);
+
+ assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testConsumerGroupDescribeWithErrors() {
+ String groupId = "groupId";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .build();
+
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.sendConsumerGroupDescribe(Collections.singletonList(groupId));
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ describedGroup
+ );
+
+ assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testConsumerGroupDescribeBeforeAndAfterCommittingOffset() {
+ String consumerGroupId = "consumerGroupId";
+ int epoch = 10;
+ String memberId1 = "memberId1";
+ String memberId2 = "memberId2";
+ String topicName = "topicName";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .build();
+
+ ConsumerGroupMember.Builder memberBuilder1 = new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(topicName));
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, memberBuilder1.build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, epoch + 1));
+
+ Map<Uuid, Set<Integer>> assignmentMap = new HashMap<>();
+ assignmentMap.put(Uuid.randomUuid(), Collections.emptySet());
+ ConsumerGroupMember.Builder memberBuilder2 = new ConsumerGroupMember.Builder(memberId2);
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, memberBuilder2.build()));
+ context.replay(RecordHelpers.newTargetAssignmentRecord(consumerGroupId, memberId2, assignmentMap));
+ context.replay(RecordHelpers.newCurrentAssignmentRecord(consumerGroupId, memberBuilder2.build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, epoch + 2));
+
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> actual = context.groupMetadataManager.consumerGroupDescribe(Collections.singletonList(consumerGroupId), context.lastCommittedOffset);
+ ConsumerGroupDescribeResponseData.DescribedGroup describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(consumerGroupId)
+ .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
+ List<ConsumerGroupDescribeResponseData.DescribedGroup> expected = Collections.singletonList(
+ describedGroup
+ );
+ assertEquals(expected, actual);
+
+ // Commit the offset and test again
+ context.commit();
+
+ actual = context.groupMetadataManager.consumerGroupDescribe(Collections.singletonList(consumerGroupId), context.lastCommittedOffset);
+ describedGroup = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(consumerGroupId)
+ .setMembers(Arrays.asList(
+ memberBuilder1.build().asConsumerGroupDescribeMember(new Assignment(Collections.emptyMap())),
+ memberBuilder2.build().asConsumerGroupDescribeMember(new Assignment(assignmentMap))
+ ))
+ .setGroupState("assigning")
+ .setAssignorName("range")
+ .setGroupEpoch(epoch + 2);
+ expected = Collections.singletonList(
+ describedGroup
+ );
+ assertEquals(expected, actual);
+ }
+
@Test
public void testDescribeGroupStable() {
GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
index 13ac57bb2b0..05ce03eda81 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupMemberTest.java
@@ -17,15 +17,23 @@
package org.apache.kafka.coordinator.group.consumer;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
import org.apache.kafka.coordinator.group.generated.ConsumerGroupMemberMetadataValue;
import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.OptionalInt;
+import java.util.Set;
+import java.util.stream.Collectors;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkAssignment;
import static org.apache.kafka.coordinator.group.AssignmentTestUtil.mkTopicAssignment;
@@ -304,4 +312,80 @@ public class ConsumerGroupMemberTest {
assertEquals(mkAssignment(mkTopicAssignment(topicId2, 3, 4, 5)), member.partitionsPendingRevocation());
assertEquals(mkAssignment(mkTopicAssignment(topicId3, 6, 7, 8)), member.partitionsPendingAssignment());
}
+
+ @Test
+ public void testAsConsumerGroupDescribeMember() {
+ Uuid topicId1 = Uuid.randomUuid();
+ Uuid topicId2 = Uuid.randomUuid();
+ Uuid topicId3 = Uuid.randomUuid();
+ List<Integer> assignedPartitions = Arrays.asList(0, 1, 2);
+ int epoch = 10;
+ ConsumerGroupCurrentMemberAssignmentValue record = new ConsumerGroupCurrentMemberAssignmentValue()
+ .setMemberEpoch(epoch)
+ .setPreviousMemberEpoch(epoch - 1)
+ .setTargetMemberEpoch(epoch + 1)
+ .setAssignedPartitions(Collections.singletonList(new ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions()
+ .setTopicId(topicId1)
+ .setPartitions(assignedPartitions)))
+ .setPartitionsPendingRevocation(Collections.singletonList(new ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions()
+ .setTopicId(topicId2)
+ .setPartitions(Arrays.asList(3, 4, 5))))
+ .setPartitionsPendingAssignment(Collections.singletonList(new ConsumerGroupCurrentMemberAssignmentValue.TopicPartitions()
+ .setTopicId(topicId3)
+ .setPartitions(Arrays.asList(6, 7, 8))));
+ String memberId = Uuid.randomUuid().toString();
+ String clientId = "clientId";
+ String instanceId = "instanceId";
+ String rackId = "rackId";
+ String clientHost = "clientHost";
+ List<String> subscribedTopicNames = Arrays.asList("topic1", "topic2");
+ String subscribedTopicRegex = "topic.*";
+ Map<Uuid, Set<Integer>> assignmentMap = new HashMap<>();
+ assignmentMap.put(Uuid.randomUuid(), new HashSet<>());
+ Assignment targetAssignment = new Assignment(assignmentMap);
+ ConsumerGroupMember member = new ConsumerGroupMember.Builder(memberId)
+ .updateWith(record)
+ .setClientId(clientId)
+ .setInstanceId(instanceId)
+ .setRackId(rackId)
+ .setClientHost(clientHost)
+ .setSubscribedTopicNames(subscribedTopicNames)
+ .setSubscribedTopicRegex(subscribedTopicRegex)
+ .build();
+
+ ConsumerGroupDescribeResponseData.Member actual = member.asConsumerGroupDescribeMember(targetAssignment);
+ ConsumerGroupDescribeResponseData.Member expected = new ConsumerGroupDescribeResponseData.Member()
+ .setMemberId(memberId)
+ .setMemberEpoch(epoch)
+ .setClientId(clientId)
+ .setInstanceId(instanceId)
+ .setRackId(rackId)
+ .setClientHost(clientHost)
+ .setSubscribedTopicNames(subscribedTopicNames)
+ .setSubscribedTopicRegex(subscribedTopicRegex)
+ .setAssignment(
+ new ConsumerGroupDescribeResponseData.Assignment()
+ .setTopicPartitions(Collections.singletonList(new ConsumerGroupDescribeResponseData.TopicPartitions().setTopicId(topicId1).setPartitions(assignedPartitions)))
+ )
+ .setTargetAssignment(
+ new ConsumerGroupDescribeResponseData.Assignment()
+ .setTopicPartitions(targetAssignment.partitions().entrySet().stream().map(
+ item -> new ConsumerGroupDescribeResponseData.TopicPartitions()
+ .setTopicId(item.getKey())
+ .setPartitions(new ArrayList<>(item.getValue()))
+ ).collect(Collectors.toList()))
+ );
+
+ assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testAsConsumerGroupDescribeWithTargetAssignmentNull() {
+ ConsumerGroupMember member = new ConsumerGroupMember.Builder(Uuid.randomUuid().toString())
+ .build();
+
+ ConsumerGroupDescribeResponseData.Member consumerGroupDescribeMember = member.asConsumerGroupDescribeMember(null);
+
+ assertEquals(new ConsumerGroupDescribeResponseData.Assignment(), consumerGroupDescribeMember.targetAssignment());
+ }
}
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
index 75016cb6650..f54ad25e321 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.GroupNotEmptyException;
import org.apache.kafka.common.errors.StaleMemberEpochException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ConsumerGroupDescribeResponseData;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.coordinator.group.GroupMetadataManagerTest;
@@ -837,4 +838,38 @@ public class ConsumerGroupTest {
consumerGroup.removeMember("member2");
assertFalse(consumerGroup.isSubscribedToTopic("bar"));
}
+
+ @Test
+ public void testAsDescribedGroup() {
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ ConsumerGroup group = new ConsumerGroup(snapshotRegistry, "group-id-1", mock(GroupCoordinatorMetricsShard.class));
+ snapshotRegistry.getOrCreateSnapshot(0);
+ assertEquals(ConsumerGroup.ConsumerGroupState.EMPTY.toString(), group.stateAsString(0));
+
+ group.updateMember(new ConsumerGroupMember.Builder("member1")
+ .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setServerAssignorName("assignorName")
+ .build());
+ group.updateMember(new ConsumerGroupMember.Builder("member2")
+ .build());
+ snapshotRegistry.getOrCreateSnapshot(1);
+
+ ConsumerGroupDescribeResponseData.DescribedGroup expected = new ConsumerGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("group-id-1")
+ .setGroupState(ConsumerGroup.ConsumerGroupState.STABLE.toString())
+ .setGroupEpoch(0)
+ .setAssignmentEpoch(0)
+ .setAssignorName("assignorName")
+ .setMembers(Arrays.asList(
+ new ConsumerGroupDescribeResponseData.Member()
+ .setMemberId("member1")
+ .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .setSubscribedTopicRegex(""),
+ new ConsumerGroupDescribeResponseData.Member().setMemberId("member2")
+ .setSubscribedTopicRegex("")
+ ));
+ ConsumerGroupDescribeResponseData.DescribedGroup actual = group.asDescribedGroup(1, "");
+
+ assertEquals(expected, actual);
+ }
}