root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: add repartitionSourceTopics to Streams group description (#19561) 

This is a follow-up of this #19433  This PR aims at adding the
`repartition source topics` to the output of `--describe` for streams
groups.

Reviewers: Lucas Brutschy <lbrutschy@confluent.io>
This commit is contained in:
Alieh Saeedi 2025-04-28 11:54:53 +02:00 committed by GitHub
parent 416f8941ca
commit 443c01ca80
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 9 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
View File

@ -368,7 +368,9 @@ public class StreamsGroupCommand {
private static Set<TopicPartition> getTopicPartitions(List<StreamsGroupMemberAssignment.TaskIds> taskIds, StreamsGroupDescription description) { private static Set<TopicPartition> getTopicPartitions(List<StreamsGroupMemberAssignment.TaskIds> taskIds, StreamsGroupDescription description) {
Map<String, List<String>> allSourceTopics = new HashMap<>(); Map<String, List<String>> allSourceTopics = new HashMap<>();
for (StreamsGroupSubtopologyDescription subtopologyDescription : description.subtopologies()) { for (StreamsGroupSubtopologyDescription subtopologyDescription : description.subtopologies()) {
allSourceTopics.put(subtopologyDescription.subtopologyId(), subtopologyDescription.sourceTopics()); List<String> topics = new ArrayList<>(subtopologyDescription.sourceTopics());
topics.addAll(subtopologyDescription.repartitionSourceTopics().keySet());
allSourceTopics.put(subtopologyDescription.subtopologyId(), topics);
} }
Set<TopicPartition> topicPartitions = new HashSet<>(); Set<TopicPartition> topicPartitions = new HashSet<>();

8
tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java
View File

@ -95,7 +95,9 @@ public class DescribeStreamsGroupTest {
final List<String> expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "OFFSET-LAG"); final List<String> expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "OFFSET-LAG");
final Set<List<String>> expectedRows = Set.of( final Set<List<String>> expectedRows = Set.of(
List.of(APP_ID, INPUT_TOPIC, "0", "0"), List.of(APP_ID, INPUT_TOPIC, "0", "0"),
List.of(APP_ID, INPUT_TOPIC, "1", "0")); List.of(APP_ID, INPUT_TOPIC, "1", "0"),
List.of(APP_ID, "streams-group-command-test-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition", "0", "0"),
List.of(APP_ID, "streams-group-command-test-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition", "1", "0"));
validateDescribeOutput( validateDescribeOutput(
Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe"), expectedHeader, expectedRows, List.of()); Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe"), expectedHeader, expectedRows, List.of());
@ -109,7 +111,9 @@ public class DescribeStreamsGroupTest {
final List<String> expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "CURRENT-OFFSET", "LEADER-EPOCH", "LOG-END-OFFSET", "OFFSET-LAG"); final List<String> expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "CURRENT-OFFSET", "LEADER-EPOCH", "LOG-END-OFFSET", "OFFSET-LAG");
final Set<List<String>> expectedRows = Set.of( final Set<List<String>> expectedRows = Set.of(
List.of(APP_ID, INPUT_TOPIC, "0", "-", "-", "0", "0"), List.of(APP_ID, INPUT_TOPIC, "0", "-", "-", "0", "0"),
List.of(APP_ID, INPUT_TOPIC, "1", "-", "-", "0", "0")); List.of(APP_ID, INPUT_TOPIC, "1", "-", "-", "0", "0"),
List.of(APP_ID, "streams-group-command-test-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition", "0", "-", "-", "0", "0"),
List.of(APP_ID, "streams-group-command-test-KSTREAM-AGGREGATE-STATE-STORE-0000000003-repartition", "1", "-", "-", "0", "0"));
validateDescribeOutput( validateDescribeOutput(
Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--verbose"), expectedHeader, expectedRows, List.of()); Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--verbose"), expectedHeader, expectedRows, List.of());