You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2025/09/23 11:06:27 UTC
[PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
lucasbru opened a new pull request, #20574:
URL: https://github.com/apache/kafka/pull/20574
When toology not configured.
In the streams group heartbeat, we validate the topology set for the group against the topic metadata, to generate the "configured topology" which has a specific number of partitions for each topic.
In streams group describe, we use the configured topology to expose this information to the user. However, we leave the topology information as null in the streams group describe response, if the topology is not configured. This triggers an IllegalStateException in the admin client implementation.
Instead, we should expose the unconfigured topology when the configured topology is not available, which will still expose useful information.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2371934925
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -95,4 +97,42 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
.collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+
+ public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() {
+ return new StreamsGroupDescribeResponseData.Topology()
+ .setEpoch(topologyEpoch)
+ .setSubtopologies(
+ subtopologies.entrySet().stream().map(
+ entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue())
+ ).toList()
Review Comment:
The subtopologies should be sorted by subtopology ID to ensure consistent ordering in the response, similar to how the test expects sorted subtopologies.
```suggestion
subtopologies.entrySet().stream()
.sorted(Map.Entry.comparingByKey())
.map(entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue()))
.toList()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20574:
URL: https://github.com/apache/kafka/pull/20574#issuecomment-3336574823
`GroupMetadataManagerTest` failed on both builds. Sound related to this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2378492322
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -95,4 +97,43 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
.collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+
+ public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() {
+ return new StreamsGroupDescribeResponseData.Topology()
+ .setEpoch(topologyEpoch)
+ .setSubtopologies(
+ subtopologies.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .map(entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue()))
+ .toList()
+ );
+ }
+
+ private StreamsGroupDescribeResponseData.Subtopology asStreamsGroupDescribeSubtopology(String subtopologyId, StreamsGroupTopologyValue.Subtopology subtopology) {
+ return new StreamsGroupDescribeResponseData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setSourceTopics(subtopology.sourceTopics().stream().sorted().toList())
+ .setRepartitionSinkTopics(subtopology.repartitionSinkTopics().stream().sorted().toList())
+ .setRepartitionSourceTopics(subtopology.repartitionSourceTopics().stream()
+ .map(this::asStreamsGroupDescribeTopicInfo)
+ .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList())
+ .setStateChangelogTopics(subtopology.stateChangelogTopics().stream()
+ .map(this::asStreamsGroupDescribeTopicInfo)
+ .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList());
+ }
+
+ private StreamsGroupDescribeResponseData.TopicInfo asStreamsGroupDescribeTopicInfo(StreamsGroupTopologyValue.TopicInfo topicInfo) {
+ return new StreamsGroupDescribeResponseData.TopicInfo()
+ .setName(topicInfo.name())
+ .setPartitions(topicInfo.partitions())
+ .setReplicationFactor(topicInfo.replicationFactor())
+ .setTopicConfigs(
+ topicInfo.topicConfigs().stream().map(
+ y -> new StreamsGroupDescribeResponseData.KeyValue()
Review Comment:
Done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1039,7 +1039,16 @@ public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset))
- .setTopology(configuredTopology.get(committedOffset).map(ConfiguredTopology::asStreamsGroupDescribeTopology).orElse(null));
+ .setTopology(
+ configuredTopology.get(committedOffset)
+ .filter(ConfiguredTopology::isReady)
+ .map(ConfiguredTopology::asStreamsGroupDescribeTopology)
+ .orElse(
+ topology.get(committedOffset)
+ .map(StreamsTopology::asStreamsGroupDescribeTopology)
+ .orElse(null)
Review Comment:
Okay, fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #20574:
URL: https://github.com/apache/kafka/pull/20574
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2377057496
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1039,7 +1039,16 @@ public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset))
- .setTopology(configuredTopology.get(committedOffset).map(ConfiguredTopology::asStreamsGroupDescribeTopology).orElse(null));
+ .setTopology(
+ configuredTopology.get(committedOffset)
+ .filter(ConfiguredTopology::isReady)
+ .map(ConfiguredTopology::asStreamsGroupDescribeTopology)
+ .orElse(
+ topology.get(committedOffset)
+ .map(StreamsTopology::asStreamsGroupDescribeTopology)
+ .orElse(null)
Review Comment:
> But I'm not sure if I want to throw an IllegalStateException here.
That's basically my question. If we just return `null`, and it's bug, it might be difficult to find out, where the bug originates from, while throwing an exception would provide a clear signal.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2374720410
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4463,6 +4463,47 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ @Test
+ def testDescribeStreamsGroupsNotReady(): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ val streams = createStreamsGroup(
+ inputTopic = testTopicName,
+ streamsGroupId = streamsGroupId
+ )
+ streams.poll(JDuration.ofMillis(500L))
+
+ try {
+ TestUtils.waitUntilTrue(() => {
+ val firstGroup = client.listGroups().all().get().stream()
+ .filter(g => g.groupId() == streamsGroupId).findFirst().orElse(null)
+ firstGroup.groupState().orElse(null) == GroupState.NOT_READY && firstGroup.groupId() == streamsGroupId
+ }, "Stream group not NOT_READY yet")
Review Comment:
The input topic does not exist, so it is guaranteed to stay in NOT_READY.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2373691400
##########
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala:
##########
@@ -4463,6 +4463,47 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
}
}
+ @Test
+ def testDescribeStreamsGroupsNotReady(): Unit = {
+ val streamsGroupId = "stream_group_id"
+ val testTopicName = "test_topic"
+
+ val config = createConfig
+ client = Admin.create(config)
+
+ val streams = createStreamsGroup(
+ inputTopic = testTopicName,
+ streamsGroupId = streamsGroupId
+ )
+ streams.poll(JDuration.ofMillis(500L))
+
+ try {
+ TestUtils.waitUntilTrue(() => {
+ val firstGroup = client.listGroups().all().get().stream()
+ .filter(g => g.groupId() == streamsGroupId).findFirst().orElse(null)
+ firstGroup.groupState().orElse(null) == GroupState.NOT_READY && firstGroup.groupId() == streamsGroupId
+ }, "Stream group not NOT_READY yet")
Review Comment:
How does this test ensure, that the group stays in NOT_READY state (ie, does not go into ASSIGNING, RECONCILING, or STABLE) ?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1039,7 +1039,16 @@ public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset))
- .setTopology(configuredTopology.get(committedOffset).map(ConfiguredTopology::asStreamsGroupDescribeTopology).orElse(null));
+ .setTopology(
+ configuredTopology.get(committedOffset)
+ .filter(ConfiguredTopology::isReady)
+ .map(ConfiguredTopology::asStreamsGroupDescribeTopology)
+ .orElse(
+ topology.get(committedOffset)
+ .map(StreamsTopology::asStreamsGroupDescribeTopology)
+ .orElse(null)
Review Comment:
Is `orElse(null)` a valid case? Should we not always have a topology, even if it's not configured yet?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -95,4 +97,43 @@ public static StreamsTopology fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
.collect(Collectors.toMap(StreamsGroupTopologyValue.Subtopology::subtopologyId, x -> x));
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+
+ public StreamsGroupDescribeResponseData.Topology asStreamsGroupDescribeTopology() {
+ return new StreamsGroupDescribeResponseData.Topology()
+ .setEpoch(topologyEpoch)
+ .setSubtopologies(
+ subtopologies.entrySet().stream()
+ .sorted(Map.Entry.comparingByKey())
+ .map(entry -> asStreamsGroupDescribeSubtopology(entry.getKey(), entry.getValue()))
+ .toList()
+ );
+ }
+
+ private StreamsGroupDescribeResponseData.Subtopology asStreamsGroupDescribeSubtopology(String subtopologyId, StreamsGroupTopologyValue.Subtopology subtopology) {
+ return new StreamsGroupDescribeResponseData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setSourceTopics(subtopology.sourceTopics().stream().sorted().toList())
+ .setRepartitionSinkTopics(subtopology.repartitionSinkTopics().stream().sorted().toList())
+ .setRepartitionSourceTopics(subtopology.repartitionSourceTopics().stream()
+ .map(this::asStreamsGroupDescribeTopicInfo)
+ .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList())
+ .setStateChangelogTopics(subtopology.stateChangelogTopics().stream()
+ .map(this::asStreamsGroupDescribeTopicInfo)
+ .sorted(Comparator.comparing(StreamsGroupDescribeResponseData.TopicInfo::name)).toList());
+ }
+
+ private StreamsGroupDescribeResponseData.TopicInfo asStreamsGroupDescribeTopicInfo(StreamsGroupTopologyValue.TopicInfo topicInfo) {
+ return new StreamsGroupDescribeResponseData.TopicInfo()
+ .setName(topicInfo.name())
+ .setPartitions(topicInfo.partitions())
+ .setReplicationFactor(topicInfo.replicationFactor())
+ .setTopicConfigs(
+ topicInfo.topicConfigs().stream().map(
+ y -> new StreamsGroupDescribeResponseData.KeyValue()
Review Comment:
Why `y` ? Can't we find a better name?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19730: StreamsGroupDescribe result is missing topology [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20574:
URL: https://github.com/apache/kafka/pull/20574#discussion_r2374717678
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -1039,7 +1039,16 @@ public StreamsGroupDescribeResponseData.DescribedGroup asDescribedGroup(
.setGroupEpoch(groupEpoch.get(committedOffset))
.setGroupState(state.get(committedOffset).toString())
.setAssignmentEpoch(targetAssignmentEpoch.get(committedOffset))
- .setTopology(configuredTopology.get(committedOffset).map(ConfiguredTopology::asStreamsGroupDescribeTopology).orElse(null));
+ .setTopology(
+ configuredTopology.get(committedOffset)
+ .filter(ConfiguredTopology::isReady)
+ .map(ConfiguredTopology::asStreamsGroupDescribeTopology)
+ .orElse(
+ topology.get(committedOffset)
+ .map(StreamsTopology::asStreamsGroupDescribeTopology)
+ .orElse(null)
Review Comment:
I don't think it's a valid case in the current code base. It could only happen if something is fundamentally wrong, for example, somehow the StreamsGroupTopologyRecord was removed from the offset commit topic. But I'm not sure if I want to throw an IllegalStateException here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org