This is an automated email from the ASF dual-hosted git repository.
lucasbru 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 12c81a2e76e KAFKA-20624: Wire topology description plugin into
GroupCoordinatorService — read path (describe) and broker wiring (#22627)
12c81a2e76e is described below
commit 12c81a2e76e4f7d78cfc7b69d9ba3ff6cb2276a7
Author: TengYao Chi <[email protected]>
AuthorDate: Mon Jun 22 13:27:12 2026 +0100
KAFKA-20624: Wire topology description plugin into GroupCoordinatorService
— read path (describe) and broker wiring (#22627)
JIRA: KAFKA-20624 This PR is a part of KIP-1331
Update RPC KafkaApis wiring. handleStreamsGroupTopologyDescriptionUpdate
has lived as a placeholder stub since KAFKA-20620 (always returning
UNSUPPORTED_VERSION), leaving the RPC unreachable end-to-end even though
the coordinator-side method merged with KAFKA-20623 split 2 (#22552).
Replace the stub with the real handler: streams-protocol-enabled gate,
READ on GROUP authorization, then dispatch to
groupCoordinator.streamsGroupTopologyDescriptionUpdate and surface
either the coordinator's response or a getErrorResponse-wrapped
exception via sendMaybeThrottle. Same shape as
handleStreamsGroupHeartbeat. The READ-on-GROUP choice follows
KIP-1331's explicit "like an offset commit, a topology push is not a
modification of the GROUP" framing — apps deployed with READ-only group
ACLs can push topologies without an ACL upgrade.
Describe path. streamsGroupDescribe is extended with an
IncludeTopologyDescription flag. When set, a new
StreamsGroupTopologyDescriptionManager.attachTopologyDescriptions
building block calls plugin.getTopology(groupId, topologyEpoch) for
each DescribedGroup whose persisted StoredDescriptionTopologyEpoch
matches the group's current topology epoch, and populates the
response's TopologyDescription and TopologyDescriptionStatus fields.
Per-group status follows KIP-1331: NOT_REQUESTED (default; client did
not ask), NOT_STORED (no description recorded or epoch mismatched or
plugin returned null), ERROR (plugin failed), AVAILABLE (topology
attached). Chain assembly stays on GroupCoordinatorService; the manager
exposes only the per-group plugin invocation as a building block,
mirroring invokeSetTopology and invokeDeleteTopologies. A new
StreamsGroupTopologyDescriptionConverter.toDescribeResponse sibling of
fromRequest translates the broker-side POJO into the wire schema and is
documented to assume the POJO's canonical-constructor non-null
collection invariants, with the manager's catch(Exception) folding any
contract bypass into per-group ERROR so a single malformed response
cannot poison the batch.
Broker wiring. BrokerServer's plugin instantiation (via
config.getConfiguredInstance) and RequestConvertToJson entries for
StreamsGroupTopologyDescriptionUpdate Request/Response and for the
TopologyDescription fields on the describe response were already in
place from earlier sub-tasks; no change needed.
Reviewers: Lucas Brutschy <[email protected]>
---
core/src/main/scala/kafka/server/KafkaApis.scala | 32 ++-
.../kafka/api/AuthorizerIntegrationTest.scala | 19 +-
.../scala/unit/kafka/server/KafkaApisTest.scala | 155 +++++++++++++-
.../kafka/coordinator/group/GroupCoordinator.java | 13 +-
.../coordinator/group/GroupCoordinatorService.java | 11 +-
.../StreamsGroupTopologyDescriptionConverter.java | 72 +++++++
.../StreamsGroupTopologyDescriptionManager.java | 133 ++++++++++++
.../group/GroupCoordinatorServiceTest.java | 8 +-
...pCoordinatorServiceTopologyDescriptionTest.java | 226 +++++++++++++++++++++
9 files changed, 645 insertions(+), 24 deletions(-)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala
b/core/src/main/scala/kafka/server/KafkaApis.scala
index a3058af66c4..8e4a695baa4 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -2922,13 +2922,32 @@ class KafkaApis(val requestChannel: RequestChannel,
}
}
- // Stub handler for KIP-1331. The full handler lands in a later sub-task;
until then this
- // responds with UNSUPPORTED_VERSION so callers fail loud rather than hit
the IllegalStateException
- // default branch in handle().
def handleStreamsGroupTopologyDescriptionUpdate(request: Request):
CompletableFuture[Unit] = {
val updateRequest =
request.body(classOf[StreamsGroupTopologyDescriptionUpdateRequest])
- requestHelper.sendMaybeThrottle(request,
updateRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
- CompletableFuture.completedFuture[Unit](())
+
+ if (!isStreamsGroupProtocolEnabled) {
+ // The streams group protocol is disabled on this broker, so the RPC is
unreachable
+ // even if a topology description plugin is configured.
+ requestHelper.sendMaybeThrottle(request,
updateRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else if (!authHelper.authorize(request.context, READ, GROUP,
updateRequest.data.groupId)) {
+ // Per KIP-1331: like offset commits, a topology push is not treated as
a modification
+ // of the GROUP, so READ on the GROUP resource is sufficient. This lets
apps deployed
+ // with READ-only group ACLs push topology descriptions without an ACL
upgrade.
+ requestHelper.sendMaybeThrottle(request,
updateRequest.getErrorResponse(Errors.GROUP_AUTHORIZATION_FAILED.exception))
+ CompletableFuture.completedFuture[Unit](())
+ } else {
+ groupCoordinator.streamsGroupTopologyDescriptionUpdate(
+ request.context,
+ updateRequest.data
+ ).handle[Unit] { (response, exception) =>
+ if (exception != null) {
+ requestHelper.sendMaybeThrottle(request,
updateRequest.getErrorResponse(exception))
+ } else {
+ requestHelper.sendMaybeThrottle(request, new
StreamsGroupTopologyDescriptionUpdateResponse(response))
+ }
+ }
+ }
}
def handleStreamsGroupDescribe(request: Request): CompletableFuture[Unit] = {
@@ -2956,7 +2975,8 @@ class KafkaApis(val requestChannel: RequestChannel,
groupCoordinator.streamsGroupDescribe(
request.context,
- authorizedGroups.asJava
+ authorizedGroups.asJava,
+ streamsGroupDescribeRequest.data.includeTopologyDescription
).handle[Unit] { (results, exception) =>
if (exception != null) {
requestHelper.sendMaybeThrottle(request,
streamsGroupDescribeRequest.getErrorResponse(exception))
diff --git
a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
index f21e75c0eb8..92e89d9df62 100644
--- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala
@@ -38,7 +38,7 @@ import
org.apache.kafka.common.message.JoinGroupRequestData.JoinGroupRequestProt
import org.apache.kafka.common.message.LeaveGroupRequestData.MemberIdentity
import
org.apache.kafka.common.message.ListOffsetsRequestData.{ListOffsetsPartition,
ListOffsetsTopic}
import
org.apache.kafka.common.message.OffsetForLeaderEpochRequestData.{OffsetForLeaderPartition,
OffsetForLeaderTopic, OffsetForLeaderTopicCollection}
-import org.apache.kafka.common.message.{AddOffsetsToTxnRequestData,
AlterPartitionReassignmentsRequestData, AlterReplicaLogDirsRequestData,
AlterShareGroupOffsetsRequestData, ConsumerGroupDescribeRequestData,
ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData,
CreateAclsRequestData, CreatePartitionsRequestData, CreateTopicsRequestData,
DeleteAclsRequestData, DeleteGroupsRequestData, DeleteRecordsRequestData,
DeleteShareGroupOffsetsRequestData, DeleteShareGroupStateRequ [...]
+import org.apache.kafka.common.message.{AddOffsetsToTxnRequestData,
AlterPartitionReassignmentsRequestData, AlterReplicaLogDirsRequestData,
AlterShareGroupOffsetsRequestData, ConsumerGroupDescribeRequestData,
ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData,
CreateAclsRequestData, CreatePartitionsRequestData, CreateTopicsRequestData,
DeleteAclsRequestData, DeleteGroupsRequestData, DeleteRecordsRequestData,
DeleteShareGroupOffsetsRequestData, DeleteShareGroupStateRequ [...]
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record.internal.{MemoryRecords, RecordBatch,
SimpleRecord}
@@ -232,7 +232,9 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
resp.data.errorCode)),
ApiKeys.STREAMS_GROUP_HEARTBEAT -> ((resp: StreamsGroupHeartbeatResponse)
=> Errors.forCode(resp.data.errorCode)),
ApiKeys.STREAMS_GROUP_DESCRIBE -> ((resp: StreamsGroupDescribeResponse) =>
- Errors.forCode(resp.data.groups.asScala.find(g => streamsGroup ==
g.groupId).head.errorCode))
+ Errors.forCode(resp.data.groups.asScala.find(g => streamsGroup ==
g.groupId).head.errorCode)),
+ ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE ->
+ ((resp: StreamsGroupTopologyDescriptionUpdateResponse) =>
Errors.forCode(resp.data.errorCode))
)
def findErrorForTopicId(id: Uuid, response: AbstractResponse): Errors = {
@@ -303,6 +305,7 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
ApiKeys.ALTER_SHARE_GROUP_OFFSETS -> (shareGroupReadAcl ++ topicReadAcl),
ApiKeys.STREAMS_GROUP_HEARTBEAT -> (streamsGroupReadAcl ++
topicDescribeAcl),
ApiKeys.STREAMS_GROUP_DESCRIBE -> (streamsGroupDescribeAcl ++
topicDescribeAcl),
+ ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE -> streamsGroupReadAcl,
)
private def createMetadataRequest(allowAutoTopicCreation: Boolean) = {
@@ -883,6 +886,14 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
new StreamsGroupDescribeRequestData()
.setGroupIds(List(streamsGroup).asJava)
.setIncludeAuthorizedOperations(false)).build(ApiKeys.STREAMS_GROUP_DESCRIBE.latestVersion)
+
+ private def streamsGroupTopologyDescriptionUpdateRequest = new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(
+ new StreamsGroupTopologyDescriptionUpdateRequestData()
+ .setGroupId(streamsGroup)
+ .setMemberId(Uuid.randomUuid.toString)
+ .setTopologyEpoch(0)
+ .setTopologyDescription(new
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescription())
+ ).build(ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE.latestVersion)
private def sendRequests(requestKeyToRequest: mutable.Map[ApiKeys,
AbstractRequest], topicExists: Boolean = true,
topicNames: Map[Uuid, String] = getTopicNames()) = {
@@ -969,6 +980,7 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
ApiKeys.ALTER_SHARE_GROUP_OFFSETS -> alterShareGroupOffsetsRequest,
ApiKeys.STREAMS_GROUP_HEARTBEAT -> streamsGroupHeartbeatRequest,
ApiKeys.STREAMS_GROUP_DESCRIBE -> streamsGroupDescribeRequest,
+ ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE ->
streamsGroupTopologyDescriptionUpdateRequest,
// Delete the topic last
ApiKeys.DELETE_TOPICS -> deleteTopicsRequest
@@ -1003,7 +1015,8 @@ class AuthorizerIntegrationTest extends
AbstractAuthorizerIntegrationTest {
ApiKeys.SHARE_ACKNOWLEDGE -> shareAcknowledgeRequest,
ApiKeys.DESCRIBE_SHARE_GROUP_OFFSETS -> describeShareGroupOffsetsRequest,
ApiKeys.STREAMS_GROUP_HEARTBEAT -> streamsGroupHeartbeatRequest,
- ApiKeys.STREAMS_GROUP_DESCRIBE -> streamsGroupDescribeRequest
+ ApiKeys.STREAMS_GROUP_DESCRIBE -> streamsGroupDescribeRequest,
+ ApiKeys.STREAMS_GROUP_TOPOLOGY_DESCRIPTION_UPDATE ->
streamsGroupTopologyDescriptionUpdateRequest
)
sendRequests(requestKeyToRequest, topicExists = false, topicNames)
diff --git a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
index e9a196118f2..de7ed5c902d 100644
--- a/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/KafkaApisTest.scala
@@ -11239,6 +11239,113 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code,
response.data.errorCode)
}
+ @Test
+ def
testStreamsGroupTopologyDescriptionUpdateReturnsUnsupportedVersionWhenStreamsProtocolDisabled():
Unit = {
+ // Streams group protocol disabled on this broker -> the gate at the top of
+ // handleStreamsGroupTopologyDescriptionUpdate short-circuits with
UNSUPPORTED_VERSION,
+ // matching the sibling handleStreamsGroupHeartbeat behavior and the
placeholder semantics
+ // KAFKA-20620 introduced before this real handler landed.
+ val updateRequest = new
StreamsGroupTopologyDescriptionUpdateRequestData().setGroupId("group")
+
+ val requestChannelRequest = buildRequest(new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(updateRequest).build())
+ metadataCache = {
+ val cache = new KRaftMetadataCache(brokerId, () =>
KRaftVersion.KRAFT_VERSION_1)
+ val delta = new MetadataDelta.Builder()
+ .setImage(MetadataImage.EMPTY)
+ .build()
+ delta.replay(new FeatureLevelRecord()
+ .setName(MetadataVersion.FEATURE_NAME)
+ .setFeatureLevel(MetadataVersion.MINIMUM_VERSION.featureLevel())
+ )
+ cache.setImage(delta.apply(MetadataProvenance.EMPTY))
+ cache
+ }
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val response =
verifyNoThrottling[StreamsGroupTopologyDescriptionUpdateResponse](requestChannelRequest)
+ assertEquals(Errors.UNSUPPORTED_VERSION.code, response.data.errorCode)
+ }
+
+ @Test
+ def testStreamsGroupTopologyDescriptionUpdateRequest(): Unit = {
+ // Happy path: streams protocol enabled, no authorizer configured (default
ALLOW),
+ // coordinator returns a clean response that must be forwarded to the
client verbatim.
+ val features = mock(classOf[FinalizedFeatures])
+
when(features.finalizedFeatures()).thenReturn(util.Map.of(StreamsVersion.FEATURE_NAME,
1.toShort))
+
+ metadataCache = mock(classOf[KRaftMetadataCache])
+ when(metadataCache.features()).thenReturn(features)
+
+ val updateRequest = new
StreamsGroupTopologyDescriptionUpdateRequestData().setGroupId("group")
+ val requestChannelRequest = buildRequest(new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(updateRequest).build())
+
+ val future = new
CompletableFuture[StreamsGroupTopologyDescriptionUpdateResponseData]()
+ when(groupCoordinator.streamsGroupTopologyDescriptionUpdate(
+ requestChannelRequest.context,
+ updateRequest
+ )).thenReturn(future)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val expectedResponse = new
StreamsGroupTopologyDescriptionUpdateResponseData()
+ future.complete(expectedResponse)
+ val response =
verifyNoThrottling[StreamsGroupTopologyDescriptionUpdateResponse](requestChannelRequest)
+ assertEquals(expectedResponse, response.data)
+ }
+
+ @Test
+ def testStreamsGroupTopologyDescriptionUpdateRequestFutureFailed(): Unit = {
+ // Coordinator-side exception (e.g. NOT_COORDINATOR, fenced member, plugin
transient
+ // failure) must be folded into an error response via getErrorResponse and
surfaced to
+ // the client.
+ val features = mock(classOf[FinalizedFeatures])
+
when(features.finalizedFeatures()).thenReturn(util.Map.of(StreamsVersion.FEATURE_NAME,
1.toShort))
+
+ metadataCache = mock(classOf[KRaftMetadataCache])
+ when(metadataCache.features()).thenReturn(features)
+
+ val updateRequest = new
StreamsGroupTopologyDescriptionUpdateRequestData().setGroupId("group")
+ val requestChannelRequest = buildRequest(new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(updateRequest).build())
+
+ val future = new
CompletableFuture[StreamsGroupTopologyDescriptionUpdateResponseData]()
+ when(groupCoordinator.streamsGroupTopologyDescriptionUpdate(
+ requestChannelRequest.context,
+ updateRequest
+ )).thenReturn(future)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ future.completeExceptionally(Errors.NOT_COORDINATOR.exception)
+ val response =
verifyNoThrottling[StreamsGroupTopologyDescriptionUpdateResponse](requestChannelRequest)
+ assertEquals(Errors.NOT_COORDINATOR.code, response.data.errorCode)
+ }
+
+ @Test
+ def
testStreamsGroupTopologyDescriptionUpdateRequestGroupAuthorizationFailed():
Unit = {
+ // Per KIP-1331 the RPC requires READ on the GROUP resource (like offset
commits). When
+ // the authorizer denies that ACL, the request must NOT reach the
coordinator — fail
+ // fast at the KafkaApis layer with GROUP_AUTHORIZATION_FAILED.
+ val features = mock(classOf[FinalizedFeatures])
+
when(features.finalizedFeatures()).thenReturn(util.Map.of(StreamsVersion.FEATURE_NAME,
1.toShort))
+
+ metadataCache = mock(classOf[KRaftMetadataCache])
+ when(metadataCache.features()).thenReturn(features)
+
+ val updateRequest = new
StreamsGroupTopologyDescriptionUpdateRequestData().setGroupId("group")
+ val requestChannelRequest = buildRequest(new
StreamsGroupTopologyDescriptionUpdateRequest.Builder(updateRequest).build())
+
+ val authorizer: Authorizer = mock(classOf[Authorizer])
+ when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
+ .thenReturn(util.List.of(AuthorizationResult.DENIED))
+ kafkaApis = createKafkaApis(authorizer = Some(authorizer))
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ val response =
verifyNoThrottling[StreamsGroupTopologyDescriptionUpdateResponse](requestChannelRequest)
+ assertEquals(Errors.GROUP_AUTHORIZATION_FAILED.code,
response.data.errorCode)
+ verify(groupCoordinator,
never()).streamsGroupTopologyDescriptionUpdate(any(), any())
+ }
+
@Test
def testStreamsGroupHeartbeatRequestTopicAuthorizationFailed(): Unit = {
val features = mock(classOf[FinalizedFeatures])
@@ -11727,7 +11834,8 @@ class KafkaApisTest extends Logging {
val future = new
CompletableFuture[util.List[StreamsGroupDescribeResponseData.DescribedGroup]]()
when(groupCoordinator.streamsGroupDescribe(
any[RequestContext],
- any[util.List[String]]
+ any[util.List[String]],
+ any[Boolean]
)).thenReturn(future)
kafkaApis = createKafkaApis()
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
@@ -11839,7 +11947,8 @@ class KafkaApisTest extends Logging {
val future = new
CompletableFuture[util.List[StreamsGroupDescribeResponseData.DescribedGroup]]()
when(groupCoordinator.streamsGroupDescribe(
any[RequestContext],
- any[util.List[String]]
+ any[util.List[String]],
+ any[Boolean]
)).thenReturn(future)
future.complete(util.List.of)
kafkaApis = createKafkaApis(
@@ -11866,7 +11975,8 @@ class KafkaApisTest extends Logging {
val future = new
CompletableFuture[util.List[StreamsGroupDescribeResponseData.DescribedGroup]]()
when(groupCoordinator.streamsGroupDescribe(
any[RequestContext],
- any[util.List[String]]
+ any[util.List[String]],
+ any[Boolean]
)).thenReturn(future)
kafkaApis = createKafkaApis()
kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
@@ -11876,6 +11986,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.FENCED_MEMBER_EPOCH.code,
response.data.groups.get(0).errorCode)
}
+ @ParameterizedTest
+ @ValueSource(booleans = Array(true, false))
+ def
testStreamsGroupDescribeForwardsIncludeTopologyDescriptionFlag(includeTopologyDescription:
Boolean): Unit = {
+ // The KafkaApis layer must thread IncludeTopologyDescription from the
request schema
+ // through to the coordinator unchanged — the coordinator gates the plugin
call on it,
+ // so a silent drop here would make the flag unobservable to operators.
+ val features = mock(classOf[FinalizedFeatures])
+
when(features.finalizedFeatures()).thenReturn(util.Map.of(StreamsVersion.FEATURE_NAME,
1.toShort))
+
+ metadataCache = mock(classOf[KRaftMetadataCache])
+ when(metadataCache.features()).thenReturn(features)
+
+ val streamsGroupDescribeRequestData = new StreamsGroupDescribeRequestData()
+ .setIncludeTopologyDescription(includeTopologyDescription)
+ streamsGroupDescribeRequestData.groupIds.add("group-id")
+ // IncludeTopologyDescription is v1+ and v1 is marked
latestVersionUnstable; build at v1
+ // explicitly so the field reaches the wire.
+ val requestChannelRequest = buildRequest(new
StreamsGroupDescribeRequest.Builder(streamsGroupDescribeRequestData).build(1.toShort))
+
+ val future = new
CompletableFuture[util.List[StreamsGroupDescribeResponseData.DescribedGroup]]()
+ when(groupCoordinator.streamsGroupDescribe(
+ any[RequestContext],
+ any[util.List[String]],
+ ArgumentMatchers.eq(includeTopologyDescription)
+ )).thenReturn(future)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handle(requestChannelRequest, RequestLocal.noCaching)
+
+ future.complete(util.List.of(new
StreamsGroupDescribeResponseData.DescribedGroup().setGroupId("group-id")))
+ verify(groupCoordinator).streamsGroupDescribe(
+ any[RequestContext],
+ any[util.List[String]],
+ ArgumentMatchers.eq(includeTopologyDescription)
+ )
+ }
+
@ParameterizedTest
@ValueSource(booleans = Array(true, false))
def
testStreamsGroupDescribeFilterUnauthorizedTopics(includeAuthorizedOperations:
Boolean): Unit = {
@@ -11916,7 +12062,8 @@ class KafkaApisTest extends Logging {
val future = new
CompletableFuture[util.List[StreamsGroupDescribeResponseData.DescribedGroup]]()
when(groupCoordinator.streamsGroupDescribe(
any[RequestContext],
- any[util.List[String]]
+ any[util.List[String]],
+ any[Boolean]
)).thenReturn(future)
kafkaApis = createKafkaApis(
authorizer = Some(authorizer)
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 717664a5939..f5995e1328b 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
@@ -240,14 +240,21 @@ public interface GroupCoordinator {
/**
* Describe streams groups.
*
- * @param context The coordinator request context.
- * @param groupIds The group ids.
+ * @param context The coordinator request context.
+ * @param groupIds The group ids.
+ * @param includeTopologyDescription Whether the client requested the
full topology
+ * description from the topology
description plugin.
+ * When {@code false}, the
+ * {@code TopologyDescription} /
{@code TopologyDescriptionStatus}
+ * fields are left at their defaults
and the plugin
+ * is not consulted.
*
* @return A future yielding the results or an exception.
*/
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>
streamsGroupDescribe(
AuthorizableRequestContext context,
- List<String> groupIds
+ List<String> groupIds,
+ boolean includeTopologyDescription
);
/**
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 80a3f29ff61..cad84307b08 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
@@ -103,7 +103,6 @@ import
org.apache.kafka.coordinator.group.GroupCoordinatorShard.DeletedTopic;
import
org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
-import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionConverter;
import
org.apache.kafka.coordinator.group.streams.StreamsGroupTopologyDescriptionManager;
@@ -419,6 +418,7 @@ public class GroupCoordinatorService implements
GroupCoordinator {
.collect(Collectors.toSet());
this.partitionMetadataClient = partitionMetadataClient;
this.streamsGroupTopologyDescriptionManager = new
StreamsGroupTopologyDescriptionManager(
+ logContext,
streamsGroupTopologyDescriptionPlugin,
time
);
@@ -1330,12 +1330,13 @@ public class GroupCoordinatorService implements
GroupCoordinator {
}
/**
- * See {@link
GroupCoordinator#streamsGroupDescribe(AuthorizableRequestContext, List)}.
+ * See {@link
GroupCoordinator#streamsGroupDescribe(AuthorizableRequestContext, List,
boolean)}.
*/
@Override
public
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>
streamsGroupDescribe(
AuthorizableRequestContext context,
- List<String> groupIds
+ List<String> groupIds,
+ boolean includeTopologyDescription
) {
if (!isActive.get()) {
return
CompletableFuture.completedFuture(StreamsGroupDescribeRequest.getErrorDescribedGroupList(
@@ -1367,7 +1368,9 @@ public class GroupCoordinatorService implements
GroupCoordinator {
"streams-group-describe",
topicPartition,
(coordinator, lastCommittedOffset) ->
coordinator.streamsGroupDescribe(groupList, lastCommittedOffset)
- ).thenApply(StreamsGroupDescribeResult::describedGroups)
+ ).thenCompose(result -> includeTopologyDescription
+ ?
streamsGroupTopologyDescriptionManager.attachTopologyDescriptions(result)
+ :
CompletableFuture.completedFuture(result.describedGroups()))
.exceptionally(exception -> handleOperationException(
"streams-group-describe",
groupList,
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
index 94de07f38b1..b5867e07c93 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionConverter.java
@@ -17,6 +17,7 @@
package org.apache.kafka.coordinator.group.streams;
import org.apache.kafka.common.errors.InvalidRequestException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.GlobalStore;
@@ -26,6 +27,7 @@ import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescri
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Source;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription.Subtopology;
+import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Optional;
@@ -81,6 +83,76 @@ public final class StreamsGroupTopologyDescriptionConverter {
return new GlobalStore((Source) source, (Processor) processor);
}
+ /**
+ * Sibling of {@link #fromRequest}: translates the broker-side {@link
StreamsGroupTopologyDescription}
+ * POJO returned by {@code plugin.getTopology} into the describe-response
wire schema. The two
+ * schemas share field names but live in different generated message
classes; they are kept in
+ * sync by this converter.
+ *
+ * <p>This method assumes every collection on the POJO (subtopologies,
nodes, successors,
+ * topics, stores, etc.) is non-null. The {@link
StreamsGroupTopologyDescription} record and
+ * its nested types enforce that invariant in their canonical constructors
via
+ * {@code Objects.requireNonNull} + {@code List.copyOf} / {@code
Collections.unmodifiableSet},
+ * so a well-formed plugin response can never reach this method with a
null collection. A
+ * pathological plugin that bypasses the constructor invariant would
surface as an
+ * {@code NullPointerException} here; that is caught at the only call site
+ * ({@code
StreamsGroupTopologyDescriptionManager#attachTopologyDescriptions}) and folded
+ * into a per-group {@code TOPOLOGY_DESCRIPTION_STATUS_ERROR}, so the rest
of the describe
+ * batch is unaffected.
+ */
+ public static StreamsGroupDescribeResponseData.TopologyDescription
toDescribeResponse(
+ StreamsGroupTopologyDescription topology
+ ) {
+ List<StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology>
subtopologies =
+ new ArrayList<>(topology.subtopologies().size());
+ for (Subtopology subtopology : topology.subtopologies()) {
+ List<StreamsGroupDescribeResponseData.TopologyDescriptionNode>
nodes =
+ new ArrayList<>(subtopology.nodes().size());
+ for (Node node : subtopology.nodes()) {
+ nodes.add(toResponseNode(node));
+ }
+ subtopologies.add(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology()
+ .setSubtopologyId(subtopology.id())
+ .setNodes(nodes)
+ );
+ }
+ List<StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore>
globalStores =
+ new ArrayList<>(topology.globalStores().size());
+ for (GlobalStore globalStore : topology.globalStores()) {
+ globalStores.add(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore()
+ .setSource(toResponseNode(globalStore.source()))
+ .setProcessor(toResponseNode(globalStore.processor()))
+ );
+ }
+ return new StreamsGroupDescribeResponseData.TopologyDescription()
+ .setSubtopologies(subtopologies)
+ .setGlobalStores(globalStores);
+ }
+
+ private static StreamsGroupDescribeResponseData.TopologyDescriptionNode
toResponseNode(Node node) {
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode wire =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName(node.name())
+ .setSuccessors(new ArrayList<>(node.successors()));
+ if (node instanceof Source source) {
+ wire.setNodeType(NODE_TYPE_SOURCE);
+ wire.setSourceTopics(new ArrayList<>(source.topics()));
+ } else if (node instanceof Processor processor) {
+ wire.setNodeType(NODE_TYPE_PROCESSOR);
+ wire.setStores(new ArrayList<>(processor.stores()));
+ } else if (node instanceof Sink sink) {
+ wire.setNodeType(NODE_TYPE_SINK);
+ sink.topic().ifPresent(wire::setSinkTopic);
+ } else {
+ throw new IllegalStateException(
+ "Unknown topology node type: " + node.getClass().getName()
+ );
+ }
+ return wire;
+ }
+
private static Node convertNode(
StreamsGroupTopologyDescriptionUpdateRequestData.TopologyDescriptionNode wire
) {
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
index 93cf81504e3..61da755aa9e 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroupTopologyDescriptionManager.java
@@ -20,16 +20,20 @@ import
org.apache.kafka.common.errors.CoordinatorLoadInProgressException;
import org.apache.kafka.common.errors.CoordinatorNotAvailableException;
import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.NotCoordinatorException;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.ApiError;
import org.apache.kafka.common.requests.StreamsGroupHeartbeatResponse.Status;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.internals.LogContext;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
+import org.slf4j.Logger;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -40,6 +44,10 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
+import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE;
+import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_ERROR;
+import static
org.apache.kafka.common.requests.StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED;
+
/**
* Broker-level component that owns the streams-group topology description
plugin
* reference and the per-group re-solicitation back-off. The chain that drives
a push
@@ -56,13 +64,16 @@ import java.util.concurrent.CompletionException;
* convergence after a restart.
*/
public class StreamsGroupTopologyDescriptionManager implements AutoCloseable {
+ private final Logger log;
private final Optional<StreamsGroupTopologyDescriptionPlugin> plugin;
private final StreamsGroupTopologyDescriptionBackoff backoff;
public StreamsGroupTopologyDescriptionManager(
+ LogContext logContext,
Optional<StreamsGroupTopologyDescriptionPlugin> plugin,
Time time
) {
+ this.log =
logContext.logger(StreamsGroupTopologyDescriptionManager.class);
this.plugin = plugin;
this.backoff = new StreamsGroupTopologyDescriptionBackoff(time);
}
@@ -299,6 +310,128 @@ public class StreamsGroupTopologyDescriptionManager
implements AutoCloseable {
"Topology description plugin failed to delete the topology."));
}
+ /**
+ * Populate {@code TopologyDescription} and {@code
TopologyDescriptionStatus} on each
+ * {@code DescribedGroup} carried by {@code result}, calling {@code
plugin.getTopology} only
+ * for groups whose persisted {@code StoredDescriptionTopologyEpoch}
matches the group's
+ * current topology epoch. Returns a future that completes when every
per-group plugin call
+ * has settled; the future never completes exceptionally — per-group
errors fold into
+ * {@code TOPOLOGY_DESCRIPTION_STATUS_ERROR} on the corresponding
describedGroup.
+ *
+ * <p>Per-group status decisions:
+ * <ul>
+ * <li>{@code errorCode != NONE} — the group could not be resolved;
leave the status
+ * field at its {@code NOT_REQUESTED} default since the client
should consult the
+ * group's error code first.</li>
+ * <li>No plugin configured on this broker — every successful group
becomes
+ * {@code NOT_STORED}: from the client's perspective the broker
simply has no
+ * description to serve.</li>
+ * <li>{@code Topology} field is null on the response (group has not yet
declared a
+ * topology), {@code storedEpoch} is missing or {@code -1}, or
{@code storedEpoch}
+ * does not match {@code topology().epoch()} — {@code
NOT_STORED}.</li>
+ * <li>Plugin call returns null (the plugin no longer has the data, e.g.
backend wipe)
+ * — {@code NOT_STORED}.</li>
+ * <li>Plugin call completes exceptionally, throws synchronously, or
returns a null
+ * future (SPI contract violation) — {@code ERROR}. Conversion
failures from the
+ * returned POJO to the wire schema also fold into {@code ERROR} so
a single
+ * malformed plugin response cannot poison the rest of the
batch.</li>
+ * <li>Plugin call returns a non-null description — {@code AVAILABLE},
with the
+ * converted topology attached.</li>
+ * </ul>
+ */
+ public
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>>
attachTopologyDescriptions(
+ StreamsGroupDescribeResult result
+ ) {
+ if (plugin.isEmpty()) {
+ for (StreamsGroupDescribeResponseData.DescribedGroup
describedGroup : result.describedGroups()) {
+ if (describedGroup.errorCode() == Errors.NONE.code()) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ }
+ }
+ return CompletableFuture.completedFuture(result.describedGroups());
+ }
+ final StreamsGroupTopologyDescriptionPlugin topologyDescriptionPlugin
= plugin.get();
+ List<CompletableFuture<Void>> pluginFutures = new ArrayList<>();
+ for (StreamsGroupDescribeResponseData.DescribedGroup describedGroup :
result.describedGroups()) {
+ CompletableFuture<Void> outcome =
maybeAttachOne(topologyDescriptionPlugin, describedGroup, result);
+ if (outcome != null) pluginFutures.add(outcome);
+ }
+ if (pluginFutures.isEmpty()) {
+ return CompletableFuture.completedFuture(result.describedGroups());
+ }
+ return CompletableFuture.allOf(pluginFutures.toArray(new
CompletableFuture<?>[0]))
+ .thenApply(unused -> result.describedGroups());
+ }
+
+ /**
+ * Inspect one describedGroup and, if eligible, fire {@code
plugin.getTopology}. Returns
+ * null when no plugin call is needed (status has already been decided
synchronously from
+ * the response shape); otherwise returns a future that completes once the
plugin call has
+ * been folded into the describedGroup's status / topology fields.
+ */
+ private CompletableFuture<Void> maybeAttachOne(
+ StreamsGroupTopologyDescriptionPlugin p,
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup,
+ StreamsGroupDescribeResult result
+ ) {
+ if (describedGroup.errorCode() != Errors.NONE.code()) {
+ return null;
+ }
+ if (describedGroup.topology() == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return null;
+ }
+ Integer storedEpoch =
result.storedDescriptionTopologyEpochs().get(describedGroup.groupId());
+ if (storedEpoch == null || storedEpoch == -1 || storedEpoch !=
describedGroup.topology().epoch()) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return null;
+ }
+ int topologyEpoch = describedGroup.topology().epoch();
+ CompletableFuture<StreamsGroupTopologyDescription> pluginFuture;
+ try {
+ pluginFuture = p.getTopology(describedGroup.groupId(),
topologyEpoch);
+ } catch (Exception e) {
+ // SPI contract violation: synchronous throw treated as ERROR.
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return CompletableFuture.completedFuture(null);
+ }
+ if (pluginFuture == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return CompletableFuture.completedFuture(null);
+ }
+ return pluginFuture.handle((topology, throwable) -> {
+ applyGetTopologyOutcome(describedGroup, topology, throwable);
+ return null;
+ });
+ }
+
+ private void applyGetTopologyOutcome(
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup,
+ StreamsGroupTopologyDescription topology,
+ Throwable throwable
+ ) {
+ if (throwable != null) {
+ Throwable cause = Errors.maybeUnwrapException(throwable);
+ log.warn("Topology description plugin getTopology failed for group
{}.",
+ describedGroup.groupId(), cause != null ? cause : throwable);
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ return;
+ }
+ if (topology == null) {
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED);
+ return;
+ }
+ try {
+ describedGroup.setTopologyDescription(
+
StreamsGroupTopologyDescriptionConverter.toDescribeResponse(topology));
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE);
+ } catch (Exception conversionError) {
+ // Defensive catch, should be unreachable in practice
+ describedGroup.setTopologyDescription(null);
+
describedGroup.setTopologyDescriptionStatus(TOPOLOGY_DESCRIPTION_STATUS_ERROR);
+ }
+ }
+
// Visible for testing.
StreamsGroupTopologyDescriptionBackoff backoff() {
return backoff;
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 33a71646068..16adc5b0618 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
@@ -2180,7 +2180,7 @@ public class GroupCoordinatorServiceTest {
)).thenReturn(describedGroupFuture);
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future
=
-
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
Arrays.asList("group-id-1", "group-id-2"));
+
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
Arrays.asList("group-id-1", "group-id-2"), false);
assertFalse(future.isDone());
describedGroupFuture.complete(new
StreamsGroupDescribeResult(List.of(describedGroup2), Map.of()));
@@ -2220,7 +2220,7 @@ public class GroupCoordinatorServiceTest {
)).thenReturn(CompletableFuture.completedFuture(List.of(describedGroup)));
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future
=
-
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
Arrays.asList("", null));
+
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
Arrays.asList("", null), false);
assertEquals(expectedDescribedGroups, future.get());
}
@@ -2244,7 +2244,7 @@ public class GroupCoordinatorServiceTest {
));
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future
=
-
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
List.of("group-id"));
+
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
List.of("group-id"), false);
assertEquals(
List.of(new StreamsGroupDescribeResponseData.DescribedGroup()
@@ -2271,7 +2271,7 @@ public class GroupCoordinatorServiceTest {
));
CompletableFuture<List<StreamsGroupDescribeResponseData.DescribedGroup>> future
=
-
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
List.of("group-id"));
+
service.streamsGroupDescribe(requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
List.of("group-id"), false);
assertEquals(
List.of(new StreamsGroupDescribeResponseData.DescribedGroup()
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
index 8fe7b89a60b..43a06050339 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTopologyDescriptionTest.java
@@ -23,20 +23,24 @@ import
org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.common.internals.Topic;
import org.apache.kafka.common.message.DeleteGroupsResponseData;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatRequestData;
import org.apache.kafka.common.message.StreamsGroupHeartbeatResponseData;
import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateRequestData;
import
org.apache.kafka.common.message.StreamsGroupTopologyDescriptionUpdateResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.StreamsGroupDescribeResponse;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.internals.BufferSupplier;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord;
import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime;
+import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescription;
import
org.apache.kafka.coordinator.group.api.streams.StreamsGroupTopologyDescriptionPlugin;
import
org.apache.kafka.coordinator.group.api.streams.StreamsTopologyDescriptionPermanentFailureException;
import org.apache.kafka.coordinator.group.metrics.GroupCoordinatorMetrics;
+import org.apache.kafka.coordinator.group.streams.StreamsGroupDescribeResult;
import org.apache.kafka.coordinator.group.streams.StreamsGroupHeartbeatResult;
import org.apache.kafka.server.share.persister.NoOpStatePersister;
import org.apache.kafka.server.util.timer.MockTimer;
@@ -913,6 +917,228 @@ public class
GroupCoordinatorServiceTopologyDescriptionTest {
assertEquals("Topology description plugin failed to delete the
topology.", badResult.errorMessage());
}
+ @Test
+ public void testDescribeWithIncludeFlagDisabledLeavesStatusDefault()
throws Exception {
+ // includeTopologyDescription=false -> plugin is not consulted
regardless of whether
+ // the group would otherwise be eligible. TopologyDescriptionStatus
stays at the
+ // default 0 (NOT_REQUESTED) and the response carries no
topologyDescription.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
describedGroupWithTopology("foo", 5);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(describedGroup),
Map.of("foo", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
false
+ ).get(5, TimeUnit.SECONDS);
+
+ assertEquals(1, result.size());
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_REQUESTED,
result.get(0).topologyDescriptionStatus());
+ assertNull(result.get(0).topologyDescription());
+ verify(plugin, never()).getTopology(anyString(), anyInt());
+ }
+
+ @Test
+ public void testDescribeSetsNotStoredWhenNoPluginConfigured() throws
Exception {
+ // includeTopologyDescription=true but no plugin on this broker: every
successful group
+ // becomes NOT_STORED (the broker has nothing to serve). The errored
group keeps its
+ // default 0 status because the client should consult the group's
errorCode first.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+
+ StreamsGroupDescribeResponseData.DescribedGroup goodGroup =
describedGroupWithTopology("good", 5);
+ StreamsGroupDescribeResponseData.DescribedGroup errorGroup = new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("error")
+ .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(goodGroup, errorGroup),
Map.of("good", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.empty(), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("good",
"error"), true
+ ).get(5, TimeUnit.SECONDS);
+
+ StreamsGroupDescribeResponseData.DescribedGroup good =
findByGroupId(result, "good");
+ StreamsGroupDescribeResponseData.DescribedGroup err =
findByGroupId(result, "error");
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED,
good.topologyDescriptionStatus());
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_REQUESTED,
err.topologyDescriptionStatus());
+ }
+
+ @Test
+ public void testDescribeFiltersNonEligibleGroupsToNotStored() throws
Exception {
+ // One test covers the three synchronous "no plugin call" branches:
topology() == null,
+ // storedEpoch missing from the result map, storedEpoch mismatched.
All three must
+ // resolve to NOT_STORED without invoking the plugin.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ StreamsGroupDescribeResponseData.DescribedGroup noTopology = new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("no-topology"); // topology field is null
+ StreamsGroupDescribeResponseData.DescribedGroup epochMissing =
describedGroupWithTopology("epoch-missing", 4);
+ StreamsGroupDescribeResponseData.DescribedGroup epochMismatch =
describedGroupWithTopology("epoch-mismatch", 4);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(
+ List.of(noTopology, epochMissing, epochMismatch),
+ // "epoch-missing" absent from map; "epoch-mismatch"
stored at 7 vs current 4.
+ Map.of("epoch-mismatch", 7))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE),
+ List.of("no-topology", "epoch-missing", "epoch-mismatch"),
+ true
+ ).get(5, TimeUnit.SECONDS);
+
+ for (StreamsGroupDescribeResponseData.DescribedGroup g : result) {
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED,
+ g.topologyDescriptionStatus(), "group " + g.groupId());
+ }
+ verify(plugin, never()).getTopology(anyString(), anyInt());
+ }
+
+ @Test
+ public void testDescribeAttachesAvailableWhenPluginReturnsTopology()
throws Exception {
+ // Happy path: storedEpoch == currentEpoch, plugin returns a non-null
topology, the
+ // wire-schema topology is attached and the status becomes AVAILABLE.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ StreamsGroupTopologyDescription pojo = new
StreamsGroupTopologyDescription(
+ List.of(new StreamsGroupTopologyDescription.Subtopology("sub-0",
List.of(
+ new StreamsGroupTopologyDescription.Source("src",
Set.of("input"), Set.of())))),
+ List.of()
+ );
+ when(plugin.getTopology("foo",
5)).thenReturn(CompletableFuture.completedFuture(pojo));
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
describedGroupWithTopology("foo", 5);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(describedGroup),
Map.of("foo", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+ StreamsGroupDescribeResponseData.DescribedGroup g = result.get(0);
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_AVAILABLE,
g.topologyDescriptionStatus());
+ assertNotNull(g.topologyDescription());
+ assertEquals(1, g.topologyDescription().subtopologies().size());
+ assertEquals("sub-0",
g.topologyDescription().subtopologies().get(0).subtopologyId());
+ }
+
+ @Test
+ public void testDescribeMarksErrorWhenPluginFails() throws Exception {
+ // Plugin future completes exceptionally -> ERROR; the converted
topology is not set.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo", 5))
+ .thenReturn(CompletableFuture.failedFuture(new
RuntimeException("plugin offline")));
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
describedGroupWithTopology("foo", 5);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(describedGroup),
Map.of("foo", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+ StreamsGroupDescribeResponseData.DescribedGroup g = result.get(0);
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_ERROR,
g.topologyDescriptionStatus());
+ assertNull(g.topologyDescription());
+ }
+
+ @Test
+ public void testDescribeMarksErrorWhenPluginThrowsSynchronously() throws
Exception {
+ // SPI contract violation: synchronous throw is treated the same as an
exceptional
+ // future -> ERROR. The exception must not propagate out of the
describe response.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo", 5)).thenThrow(new
RuntimeException("synthetic sync throw"));
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
describedGroupWithTopology("foo", 5);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(describedGroup),
Map.of("foo", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_ERROR,
+ result.get(0).topologyDescriptionStatus());
+ }
+
+ @Test
+ public void testDescribeMarksNotStoredWhenPluginReturnsNullDescription()
throws Exception {
+ // Plugin's future completes with null -> the plugin no longer holds
the data
+ // (e.g. backend wipe). Treat as NOT_STORED, not ERROR — the broker
successfully
+ // queried the plugin and learned there is nothing to return.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+ when(plugin.getTopology("foo",
5)).thenReturn(CompletableFuture.completedFuture(null));
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
describedGroupWithTopology("foo", 5);
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(describedGroup),
Map.of("foo", 5))));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("foo"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_STORED,
+ result.get(0).topologyDescriptionStatus());
+ assertNull(result.get(0).topologyDescription());
+ }
+
+ @Test
+ public void testDescribeLeavesErroredGroupsAlone() throws Exception {
+ // Groups with non-NONE errorCode are not eligible for topology
attach; status stays
+ // at the default 0 (NOT_REQUESTED). The plugin must not be called for
them.
+ CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime =
mockRuntime();
+ StreamsGroupTopologyDescriptionPlugin plugin =
mock(StreamsGroupTopologyDescriptionPlugin.class);
+
+ StreamsGroupDescribeResponseData.DescribedGroup errorGroup = new
StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId("err")
+ .setErrorCode(Errors.GROUP_ID_NOT_FOUND.code());
+ when(runtime.scheduleReadOperation(eq("streams-group-describe"),
eq(GROUP_TP), any()))
+ .thenReturn(CompletableFuture.completedFuture(
+ new StreamsGroupDescribeResult(List.of(errorGroup),
Map.of())));
+
+ GroupCoordinatorService service = buildService(runtime,
Optional.of(plugin), true);
+ List<StreamsGroupDescribeResponseData.DescribedGroup> result =
service.streamsGroupDescribe(
+ requestContext(ApiKeys.STREAMS_GROUP_DESCRIBE), List.of("err"),
true
+ ).get(5, TimeUnit.SECONDS);
+
+
assertEquals(StreamsGroupDescribeResponse.TOPOLOGY_DESCRIPTION_STATUS_NOT_REQUESTED,
+ result.get(0).topologyDescriptionStatus());
+ verify(plugin, never()).getTopology(anyString(), anyInt());
+ }
+
+ private static StreamsGroupDescribeResponseData.DescribedGroup
describedGroupWithTopology(
+ String groupId, int topologyEpoch
+ ) {
+ return new StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setErrorCode(Errors.NONE.code())
+ .setTopology(new
StreamsGroupDescribeResponseData.Topology().setEpoch(topologyEpoch));
+ }
+
+ private static StreamsGroupDescribeResponseData.DescribedGroup
findByGroupId(
+ List<StreamsGroupDescribeResponseData.DescribedGroup> groups, String
groupId
+ ) {
+ return groups.stream().filter(g ->
g.groupId().equals(groupId)).findFirst().orElseThrow();
+ }
+
private static StreamsGroupTopologyDescriptionUpdateRequestData
validUpdateRequest() {
return new StreamsGroupTopologyDescriptionUpdateRequestData()
.setGroupId("foo")
