lucasbru commented on code in PR #22636: URL: https://github.com/apache/kafka/pull/22636#discussion_r3458678184
########## 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'); Review Comment: `appendSubtopology` ends with `sb.append('\n')` to leave a blank line between subtopologies, but `appendGlobalStore` does not. Two consecutive global stores run together with no blank-line separator, diverging from `Topology#describe()` format. ########## clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescription.java: ########## @@ -0,0 +1,419 @@ +/* + * 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.clients.admin; + +import org.apache.kafka.common.annotation.InterfaceStability; + +import java.util.Collection; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.Set; + +/** + * A description of a Kafka Streams topology, as recorded by the topology description plugin configured on the broker. + * <p> + * This type mirrors {@code org.apache.kafka.streams.TopologyDescription} in shape but lives in the admin client so that + * callers do not need to depend on {@code kafka-streams}. The wire format only carries the successor relation between + * nodes; the {@link Node#predecessors() predecessors} are reconstructed from the successors when this description is built. + */ [email protected] +public class StreamsGroupTopologyDescription { + + private final Collection<Subtopology> subtopologies; + private final Collection<GlobalStore> globalStores; + + public StreamsGroupTopologyDescription( + final Collection<Subtopology> subtopologies, + final Collection<GlobalStore> globalStores + ) { + this.subtopologies = List.copyOf(Objects.requireNonNull(subtopologies, "subtopologies must be non-null")); + this.globalStores = List.copyOf(Objects.requireNonNull(globalStores, "globalStores must be non-null")); + } + + /** + * The subtopologies that make up this topology. + */ + public Collection<Subtopology> subtopologies() { + return subtopologies; + } + + /** + * The global state stores used by this topology. + */ + public Collection<GlobalStore> globalStores() { + return globalStores; + } + + @Override + public boolean equals(final Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final StreamsGroupTopologyDescription that = (StreamsGroupTopologyDescription) o; + return Objects.equals(subtopologies, that.subtopologies) + && Objects.equals(globalStores, that.globalStores); + } + + @Override + public int hashCode() { + return Objects.hash(subtopologies, globalStores); + } + + @Override + public String toString() { + return "StreamsGroupTopologyDescription(" + + "subtopologies=" + subtopologies + + ", globalStores=" + globalStores + + ')'; + } + + /** + * A connected sub-graph of a topology. + */ + public static class Subtopology { + + private final String id; + private final Collection<Node> nodes; + + public Subtopology(final String id, final Collection<Node> nodes) { + this.id = Objects.requireNonNull(id, "id must be non-null"); + this.nodes = List.copyOf(Objects.requireNonNull(nodes, "nodes must be non-null")); + } + + /** + * The subtopology identifier, unique within the topology. + */ + public String id() { + return id; + } + + /** + * The processing nodes in this subtopology. + */ + public Collection<Node> nodes() { + return nodes; + } + + @Override + public boolean equals(final Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final Subtopology that = (Subtopology) o; + return Objects.equals(id, that.id) && Objects.equals(nodes, that.nodes); Review Comment: `nodes` is stored as `List.copyOf(...)` but typed `Collection<Node>`, so `Objects.equals(nodes, that.nodes)` delegates to `List.equals` — order-sensitive. Two `Subtopology` objects with the same nodes in different wire order compare as unequal. Since nodes in a subtopology are unordered, `equals`/`hashCode` should either convert to a set or store the field as a `Set`. -- 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]
