Copilot commented on code in PR #22636:
URL: https://github.com/apache/kafka/pull/22636#discussion_r3458298087
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java:
##########
@@ -189,6 +224,84 @@ private Map<String,
StreamsGroupSubtopologyDescription.TopicInfo> convertTopicIn
));
}
+ private Optional<StreamsGroupTopologyDescription>
convertTopologyDescription(
+ final StreamsGroupTopologyDescriptionStatus status,
+ final StreamsGroupDescribeResponseData.TopologyDescription
topologyDescription) {
+ // The description is present if and only if the status is AVAILABLE.
+ if (status != StreamsGroupTopologyDescriptionStatus.AVAILABLE ||
topologyDescription == null) {
+ return Optional.empty();
+ }
Review Comment:
If the broker returns status AVAILABLE but omits the topology description
field (null), this code returns Optional.empty() while still setting
topologyDescriptionStatus=AVAILABLE. That violates the StreamsGroupDescription
contract ("present iff AVAILABLE") and can lead to downstream
NoSuchElementException (e.g., CLI uses orElseThrow()). Treat AVAILABLE+null as
a malformed response and fail the group.
##########
tools/src/main/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatter.java:
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.streams;
+
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.GlobalStore;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Node;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Processor;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Sink;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Source;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Subtopology;
+
+import java.util.Collection;
+import java.util.TreeSet;
+
+/**
+ * Formats a {@link StreamsGroupTopologyDescription} as human-readable text,
mirroring the output of
+ * {@code org.apache.kafka.streams.Topology#describe()} so that users see a
familiar representation. Node names, topics,
+ * stores and the successor/predecessor relations are sorted for stable,
readable output.
+ */
+public final class TopologyDescriptionFormatter {
+
+ private TopologyDescriptionFormatter() {
+ }
+
+ public static String format(final StreamsGroupTopologyDescription
topology) {
+ final StringBuilder sb = new StringBuilder();
+ sb.append("Topologies:\n");
+ for (final Subtopology subtopology : topology.subtopologies()) {
+ sb.append(" ");
+ appendSubtopology(sb, subtopology);
+ }
+ // The wire format does not carry global store ids, so number them
sequentially after the subtopologies.
+ int globalStoreId = topology.subtopologies().size();
+ for (final GlobalStore globalStore : topology.globalStores()) {
+ sb.append(" ");
+ appendGlobalStore(sb, globalStore, globalStoreId++);
+ }
+ return sb.toString();
+ }
+
+ private static void appendSubtopology(final StringBuilder sb, final
Subtopology subtopology) {
+ sb.append("Sub-topology: ").append(subtopology.id()).append('\n');
+ for (final Node node : subtopology.nodes()) {
+ sb.append(" ");
+ appendNode(sb, node);
+ sb.append('\n');
+ }
+ sb.append('\n');
+ }
+
+ private static void appendGlobalStore(final StringBuilder sb, final
GlobalStore globalStore, final int id) {
+ sb.append("Sub-topology: ").append(id).append(" for global store (will
not generate tasks)\n");
+ sb.append(" ");
+ appendNode(sb, globalStore.source());
+ sb.append('\n');
+ sb.append(" ");
+ appendNode(sb, globalStore.processor());
+ sb.append('\n');
+ }
+
+ private static void appendNode(final StringBuilder sb, final Node node) {
+ if (node instanceof Source) {
+ final Source source = (Source) node;
+ sb.append("Source: ").append(source.name())
+ .append(" (topics:
").append(sorted(source.topics())).append(")")
+ .append("\n --> ").append(nodeNames(source.successors()));
+ } else if (node instanceof Processor) {
+ final Processor processor = (Processor) node;
+ sb.append("Processor: ").append(processor.name())
+ .append(" (stores:
").append(sorted(processor.stores())).append(")")
+ .append("\n -->
").append(nodeNames(processor.successors()))
+ .append("\n <--
").append(nodeNames(processor.predecessors()));
+ } else if (node instanceof Sink) {
+ final Sink sink = (Sink) node;
+ sb.append("Sink: ").append(sink.name())
+ .append(" (topic:
").append(sink.topic().orElse(null)).append(")")
+ .append("\n <-- ").append(nodeNames(sink.predecessors()));
+ } else {
+ throw new IllegalStateException("Unknown topology node type: " +
node.getClass().getName());
+ }
+ }
+
+ private static TreeSet<String> sorted(final Collection<String> values) {
+ return new TreeSet<>(values);
+ }
+
+ private static String nodeNames(final Collection<String> names) {
+ return String.join(", ", sorted(names));
+ }
Review Comment:
To actually mirror Kafka Streams' `Topology#describe()` output, empty
successor/predecessor lists should be rendered as "none" (see
streams/processor/internals/InternalTopologyBuilder#nodeNames), not as an empty
string. The current implementation also leaves trailing whitespace on lines
like "--> ".
##########
tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.streams;
+
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.GlobalStore;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Node;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Processor;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Sink;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Source;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Subtopology;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TopologyDescriptionFormatterTest {
+
+ @Test
+ public void testFormatSimpleSubtopology() {
+ Source source = new Source("source", Set.of("input"),
Set.of("processor"), Set.of());
+ Processor processor = new Processor("processor", Set.of("store"),
Set.of("sink"), Set.of("source"));
+ Sink sink = new Sink("sink", Optional.of("output"), Set.of(),
Set.of("processor"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new Subtopology("0", List.<Node>of(source, processor,
sink))),
+ List.of());
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source (topics: [input])\n" +
+ " --> processor\n" +
+ " Processor: processor (stores: [store])\n" +
+ " --> sink\n" +
+ " <-- source\n" +
+ " Sink: sink (topic: output)\n" +
+ " <-- processor\n" +
+ "\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatGlobalStore() {
+ Source source = new Source("global-source", Set.of("global-topic"),
Set.of("global-processor"), Set.of());
+ Processor processor = new Processor("global-processor",
Set.of("global-store"), Set.of(), Set.of("global-source"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(),
+ List.of(new GlobalStore(source, processor)));
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0 for global store (will not generate tasks)\n" +
+ " Source: global-source (topics: [global-topic])\n" +
+ " --> global-processor\n" +
+ " Processor: global-processor (stores: [global-store])\n" +
+ " --> \n" +
+ " <-- global-source\n";
Review Comment:
This expected output should match the formatter's behavior for empty
successor lists (Kafka Streams prints "none" rather than a blank after "-->"
when there are no successors). Update the expectation to avoid asserting a
trailing whitespace-only line.
##########
tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.streams;
+
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.GlobalStore;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Node;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Processor;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Sink;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Source;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Subtopology;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TopologyDescriptionFormatterTest {
+
+ @Test
+ public void testFormatSimpleSubtopology() {
+ Source source = new Source("source", Set.of("input"),
Set.of("processor"), Set.of());
+ Processor processor = new Processor("processor", Set.of("store"),
Set.of("sink"), Set.of("source"));
+ Sink sink = new Sink("sink", Optional.of("output"), Set.of(),
Set.of("processor"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new Subtopology("0", List.<Node>of(source, processor,
sink))),
+ List.of());
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source (topics: [input])\n" +
+ " --> processor\n" +
+ " Processor: processor (stores: [store])\n" +
+ " --> sink\n" +
+ " <-- source\n" +
+ " Sink: sink (topic: output)\n" +
+ " <-- processor\n" +
+ "\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatGlobalStore() {
+ Source source = new Source("global-source", Set.of("global-topic"),
Set.of("global-processor"), Set.of());
+ Processor processor = new Processor("global-processor",
Set.of("global-store"), Set.of(), Set.of("global-source"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(),
+ List.of(new GlobalStore(source, processor)));
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0 for global store (will not generate tasks)\n" +
+ " Source: global-source (topics: [global-topic])\n" +
+ " --> global-processor\n" +
+ " Processor: global-processor (stores: [global-store])\n" +
+ " --> \n" +
+ " <-- global-source\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatMultipleSubtopologiesAreIndentedConsistently() {
+ Source source0 = new Source("source-0", Set.of("input-0"), Set.of(),
Set.of());
+ Source source1 = new Source("source-1", Set.of("input-1"), Set.of(),
Set.of());
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(
+ new Subtopology("0", List.<Node>of(source0)),
+ new Subtopology("1", List.<Node>of(source1))),
+ List.of());
+
+ // Both subtopologies are prefixed with the same three-space
indentation.
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source-0 (topics: [input-0])\n" +
+ " --> \n" +
+ "\n" +
Review Comment:
This expectation should use "--> none" for a source node with no successors
to match `TopologyDescriptionFormatter` (and Kafka Streams'
`Topology#describe()`) output.
##########
tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.streams;
+
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.GlobalStore;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Node;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Processor;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Sink;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Source;
+import
org.apache.kafka.clients.admin.StreamsGroupTopologyDescription.Subtopology;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TopologyDescriptionFormatterTest {
+
+ @Test
+ public void testFormatSimpleSubtopology() {
+ Source source = new Source("source", Set.of("input"),
Set.of("processor"), Set.of());
+ Processor processor = new Processor("processor", Set.of("store"),
Set.of("sink"), Set.of("source"));
+ Sink sink = new Sink("sink", Optional.of("output"), Set.of(),
Set.of("processor"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new Subtopology("0", List.<Node>of(source, processor,
sink))),
+ List.of());
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source (topics: [input])\n" +
+ " --> processor\n" +
+ " Processor: processor (stores: [store])\n" +
+ " --> sink\n" +
+ " <-- source\n" +
+ " Sink: sink (topic: output)\n" +
+ " <-- processor\n" +
+ "\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatGlobalStore() {
+ Source source = new Source("global-source", Set.of("global-topic"),
Set.of("global-processor"), Set.of());
+ Processor processor = new Processor("global-processor",
Set.of("global-store"), Set.of(), Set.of("global-source"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(),
+ List.of(new GlobalStore(source, processor)));
+
+ String expected = "Topologies:\n" +
+ " Sub-topology: 0 for global store (will not generate tasks)\n" +
+ " Source: global-source (topics: [global-topic])\n" +
+ " --> global-processor\n" +
+ " Processor: global-processor (stores: [global-store])\n" +
+ " --> \n" +
+ " <-- global-source\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatMultipleSubtopologiesAreIndentedConsistently() {
+ Source source0 = new Source("source-0", Set.of("input-0"), Set.of(),
Set.of());
+ Source source1 = new Source("source-1", Set.of("input-1"), Set.of(),
Set.of());
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(
+ new Subtopology("0", List.<Node>of(source0)),
+ new Subtopology("1", List.<Node>of(source1))),
+ List.of());
+
+ // Both subtopologies are prefixed with the same three-space
indentation.
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source-0 (topics: [input-0])\n" +
+ " --> \n" +
+ "\n" +
+ " Sub-topology: 1\n" +
+ " Source: source-1 (topics: [input-1])\n" +
+ " --> \n" +
+ "\n", TopologyDescriptionFormatter.format(topology));
Review Comment:
Same as above: for a source with no successors, Kafka Streams prints "-->
none" rather than leaving the arrow line blank.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
