From 443c01ca80cbeecd8c1b4272742cd5b05f43568e Mon Sep 17 00:00:00 2001 From: Alieh Saeedi <107070585+aliehsaeedii@users.noreply.github.com> Date: Mon, 28 Apr 2025 11:54:53 +0200 Subject: [PATCH] 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 --- .../apache/kafka/tools/streams/StreamsGroupCommand.java | 4 +++- .../kafka/tools/streams/DescribeStreamsGroupTest.java | 8 ++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java index bbf09c3f36a..1a8be4104aa 100644 --- a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java +++ b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java @@ -368,7 +368,9 @@ public class StreamsGroupCommand { private static Set getTopicPartitions(List taskIds, StreamsGroupDescription description) { Map> allSourceTopics = new HashMap<>(); for (StreamsGroupSubtopologyDescription subtopologyDescription : description.subtopologies()) { - allSourceTopics.put(subtopologyDescription.subtopologyId(), subtopologyDescription.sourceTopics()); + List topics = new ArrayList<>(subtopologyDescription.sourceTopics()); + topics.addAll(subtopologyDescription.repartitionSourceTopics().keySet()); + allSourceTopics.put(subtopologyDescription.subtopologyId(), topics); } Set topicPartitions = new HashSet<>(); diff --git a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java index 09efaebd0b1..6af3232dd4c 100644 --- a/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java @@ -95,7 +95,9 @@ public class DescribeStreamsGroupTest { final List expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "OFFSET-LAG"); final Set> expectedRows = Set.of( 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( Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe"), expectedHeader, expectedRows, List.of()); @@ -109,7 +111,9 @@ public class DescribeStreamsGroupTest { final List expectedHeader = List.of("GROUP", "TOPIC", "PARTITION", "CURRENT-OFFSET", "LEADER-EPOCH", "LOG-END-OFFSET", "OFFSET-LAG"); final Set> expectedRows = Set.of( 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( Arrays.asList("--bootstrap-server", cluster.bootstrapServers(), "--describe", "--verbose"), expectedHeader, expectedRows, List.of());