Copilot commented on code in PR #22636: URL: https://github.com/apache/kafka/pull/22636#discussion_r3452926814
########## 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(); + } Review Comment: `format` currently relies on the initial "Topologies:\n " (note the trailing space) plus per-item " " prefixes to get 3-space indentation. This produces inconsistent indentation for the 2nd+ subtopology/global store (only 2 spaces) and leaves a trailing space when there are no entries. Prefer emitting "Topologies:\n" and prefixing each section with a consistent " ". ########## clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java: ########## @@ -120,6 +131,11 @@ public ApiResult<CoordinatorKey, StreamsGroupDescription> handleResponse( final Set<AclOperation> authorizedOperations = validAclOperations(describedGroup.authorizedOperations()); + final StreamsGroupTopologyDescriptionStatus topologyDescriptionStatus = + StreamsGroupTopologyDescriptionStatus.forId(describedGroup.topologyDescriptionStatus()); + final Optional<StreamsGroupTopologyDescription> topologyDescription = + convertTopologyDescription(topologyDescriptionStatus, describedGroup.topologyDescription()); + Review Comment: If the broker returns `topologyDescriptionStatus=AVAILABLE` but `topologyDescription` is null, the handler currently builds a `StreamsGroupDescription` with status AVAILABLE and an empty Optional. This violates the API contract and will later crash callers (e.g. CLI uses `orElseThrow()` when status is AVAILABLE). Consider treating this as a malformed response and failing the key. ########## tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java: ########## @@ -279,16 +286,44 @@ public void describeGroups() throws ExecutionException, InterruptedException { } } } + return exitCode; } StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException { + return getDescribeGroup(group, false); + } + + StreamsGroupDescription getDescribeGroup(String group, boolean includeTopologyDescription) throws ExecutionException, InterruptedException { DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups( List.of(group), - withTimeoutMs(new DescribeStreamsGroupsOptions())); + withTimeoutMs(new DescribeStreamsGroupsOptions().includeTopologyDescription(includeTopologyDescription))); Map<String, StreamsGroupDescription> descriptionMap = result.all().get(); return descriptionMap.get(group); } + /** + * Prints the topology description for the given group. Returns {@code true} if a description was available and + * printed, {@code false} otherwise (so the caller can surface a non-zero exit code). + */ + private boolean printTopology(StreamsGroupDescription description) { + switch (description.topologyDescriptionStatus()) { + case AVAILABLE: + System.out.println(TopologyDescriptionFormatter.format(description.topologyDescription().orElseThrow())); Review Comment: `TopologyDescriptionFormatter.format(...)` already includes trailing newline(s). Using `println` here adds an extra blank line after every topology output. Use `print` to avoid double-newlines. -- 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]
