This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 402087dc42e KAFKA-20626: Add topology description to Admin client,
DescribeStreamsGroups handler, and CLI tools (#22636)
402087dc42e is described below
commit 402087dc42e8ec06751e9bbc312f4bf029db0488
Author: Alieh Saeedi <[email protected]>
AuthorDate: Tue Jun 23 21:02:56 2026 +0200
KAFKA-20626: Add topology description to Admin client,
DescribeStreamsGroups handler, and CLI tools (#22636)
## Summary
Implements the client-side surface of
[KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)
([KAFKA-20626](https://issues.apache.org/jira/browse/KAFKA-20626)), on
top of the already-defined `StreamsGroupDescribe` v1 schema
(`IncludeTopologyDescription` / `TopologyDescription` /
`TopologyDescriptionStatus`).
### Admin client
- **`StreamsGroupTopologyDescription`** — admin POJO mirroring
`org.apache.kafka.streams.TopologyDescription` (Subtopology,
Source/Processor/Sink nodes, GlobalStore) but string-based, so callers
don't depend on `kafka-streams`. The wire format only carries
successors; **predecessor edges are reconstructed** from the successor
lists when the description is built.
- **`StreamsGroupTopologyDescriptionStatus`** — enum
`NOT_REQUESTED(0)`/`NOT_STORED(1)`/`ERROR(2)`/`AVAILABLE(3)` with
`id()`/`forId()`.
- Extended **`StreamsGroupDescription`** with `topologyDescription()`
(`Optional`) and `topologyDescriptionStatus()`. The new fields are added
via an additive constructor; the existing constructor defaults to
`NOT_REQUESTED`/empty, preserving backward compatibility.
- Extended **`DescribeStreamsGroupsOptions`** with
`includeTopologyDescription(boolean)`.
- **`DescribeStreamsGroupsHandler`** sets the request flag, parses the
topology (including the global-store source→processor pair), and
populates the description only when the status is `AVAILABLE`.
`KafkaAdminClient` passes the option through.
- Implemented **`MockAdminClient.describeStreamsGroups`** (with
`addStreamsGroupDescription`).
### CLI (`kafka-streams-groups.sh`)
- New `--describe --topology` sub-action (mutually exclusive with
`--members`/`--offsets`/`--state`).
- **`TopologyDescriptionFormatter`** renders output mirroring
`Topology#describe()`, with sorted node/topic/store sets for stable
output.
- Exit code `0` for `AVAILABLE`, `1` for `NOT_STORED`/`ERROR`.
### Tests & docs
- `DescribeStreamsGroupsHandlerTest`,
`TopologyDescriptionFormatterTest`, and CLI coverage in
`StreamsGroupCommandTest` (available → exit 0 + correct output + option
captured; not-stored → exit 1).
- Documented the new option in `kafka-streams-group-sh.md`.
### Note
`StreamsGroupDescribeRequest.json` v1 remains `latestVersionUnstable:
true` (per its own comment, it flips when the broker
topology-description manager lands). Until then, `--topology` against a
live cluster surfaces `UnsupportedVersionException`, which is the KIP's
documented compatibility behavior. Unit tests construct the
request/response directly and are unaffected.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: Lucas Brutschy <[email protected]>
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../admin/DescribeStreamsGroupsOptions.java | 15 +
.../kafka/clients/admin/KafkaAdminClient.java | 2 +-
.../clients/admin/StreamsGroupDescription.java | 59 ++-
.../admin/StreamsGroupTopologyDescription.java | 422 +++++++++++++++++++++
.../StreamsGroupTopologyDescriptionStatus.java | 82 ++++
.../internals/DescribeStreamsGroupsHandler.java | 119 +++++-
.../kafka/clients/admin/MockAdminClientTest.java | 132 +++++++
.../DescribeStreamsGroupsHandlerTest.java | 311 +++++++++++++++
.../kafka/clients/admin/MockAdminClient.java | 43 ++-
.../developer-guide/kafka-streams-group-sh.md | 7 +-
.../kafka/tools/streams/StreamsGroupCommand.java | 49 ++-
.../tools/streams/StreamsGroupCommandOptions.java | 9 +-
.../streams/TopologyDescriptionFormatter.java | 108 ++++++
.../tools/streams/StreamsGroupCommandTest.java | 62 +++
.../streams/TopologyDescriptionFormatterTest.java | 167 ++++++++
15 files changed, 1570 insertions(+), 17 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeStreamsGroupsOptions.java
b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeStreamsGroupsOptions.java
index 88e6d097768..65f11f8c362 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeStreamsGroupsOptions.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeStreamsGroupsOptions.java
@@ -29,6 +29,7 @@ import java.util.Collection;
@InterfaceStability.Evolving
public class DescribeStreamsGroupsOptions extends
AbstractOptions<DescribeStreamsGroupsOptions> {
private boolean includeAuthorizedOperations;
+ private boolean includeTopologyDescription;
public DescribeStreamsGroupsOptions includeAuthorizedOperations(boolean
includeAuthorizedOperations) {
this.includeAuthorizedOperations = includeAuthorizedOperations;
@@ -38,4 +39,18 @@ public class DescribeStreamsGroupsOptions extends
AbstractOptions<DescribeStream
public boolean includeAuthorizedOperations() {
return includeAuthorizedOperations;
}
+
+ /**
+ * Whether to include the full topology description, as recorded by the
broker's topology description plugin, in the
+ * response. Requesting a topology description against a broker that does
not support it fails with
+ * {@link org.apache.kafka.common.errors.UnsupportedVersionException}.
+ */
+ public DescribeStreamsGroupsOptions includeTopologyDescription(boolean
includeTopologyDescription) {
+ this.includeTopologyDescription = includeTopologyDescription;
+ return this;
+ }
+
+ public boolean includeTopologyDescription() {
+ return includeTopologyDescription;
+ }
}
\ No newline at end of file
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index a02b4e7bcf2..5e2924e9ce9 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -3856,7 +3856,7 @@ public class KafkaAdminClient extends AdminClient {
final
DescribeStreamsGroupsOptions options) {
SimpleAdminApiFuture<CoordinatorKey, StreamsGroupDescription> future =
DescribeStreamsGroupsHandler.newFuture(groupIds);
- DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(options.includeAuthorizedOperations(), logContext);
+ DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(options.includeAuthorizedOperations(),
options.includeTopologyDescription(), logContext);
invokeDriver(handler, future, options.timeoutMs);
return new DescribeStreamsGroupsResult(future.all().entrySet().stream()
.collect(Collectors.toMap(entry -> entry.getKey().idValue,
Map.Entry::getValue)));
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
index 024285324a0..cb74803a610 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupDescription.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.annotation.InterfaceStability;
import java.util.Collection;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
@@ -42,6 +43,8 @@ public class StreamsGroupDescription {
private final GroupState groupState;
private final Node coordinator;
private final Set<AclOperation> authorizedOperations;
+ private final Optional<StreamsGroupTopologyDescription>
topologyDescription;
+ private final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus;
public StreamsGroupDescription(
final String groupId,
@@ -53,6 +56,34 @@ public class StreamsGroupDescription {
final GroupState groupState,
final Node coordinator,
final Set<AclOperation> authorizedOperations
+ ) {
+ this(
+ groupId,
+ groupEpoch,
+ targetAssignmentEpoch,
+ topologyEpoch,
+ subtopologies,
+ members,
+ groupState,
+ coordinator,
+ authorizedOperations,
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED
+ );
+ }
+
+ public StreamsGroupDescription(
+ final String groupId,
+ final int groupEpoch,
+ final int targetAssignmentEpoch,
+ final int topologyEpoch,
+ final Collection<StreamsGroupSubtopologyDescription> subtopologies,
+ final Collection<StreamsGroupMemberDescription> members,
+ final GroupState groupState,
+ final Node coordinator,
+ final Set<AclOperation> authorizedOperations,
+ final Optional<StreamsGroupTopologyDescription>
topologyDescription,
+ final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus
) {
this.groupId = Objects.requireNonNull(groupId, "groupId must be
non-null");
this.groupEpoch = groupEpoch;
@@ -63,6 +94,8 @@ public class StreamsGroupDescription {
this.groupState = Objects.requireNonNull(groupState, "groupState must
be non-null");
this.coordinator = Objects.requireNonNull(coordinator, "coordinator
must be non-null");
this.authorizedOperations = authorizedOperations;
+ this.topologyDescription = Objects.requireNonNull(topologyDescription,
"topologyDescription must be non-null");
+ this.topologyDescriptionStatus =
Objects.requireNonNull(topologyDescriptionStatus, "topologyDescriptionStatus
must be non-null");
}
/**
@@ -128,6 +161,22 @@ public class StreamsGroupDescription {
return authorizedOperations;
}
+ /**
+ * The full topology description for this group, as recorded by the
broker's topology description plugin.
+ * Present if and only if {@link #topologyDescriptionStatus()} is
+ * {@link StreamsGroupTopologyDescriptionStatus#AVAILABLE AVAILABLE}.
+ */
+ public Optional<StreamsGroupTopologyDescription> topologyDescription() {
+ return topologyDescription;
+ }
+
+ /**
+ * The status of the topology description for this group, paired with
{@link #topologyDescription()}.
+ */
+ public StreamsGroupTopologyDescriptionStatus topologyDescriptionStatus() {
+ return topologyDescriptionStatus;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) {
@@ -145,7 +194,9 @@ public class StreamsGroupDescription {
&& Objects.equals(members, that.members)
&& groupState == that.groupState
&& Objects.equals(coordinator, that.coordinator)
- && Objects.equals(authorizedOperations, that.authorizedOperations);
+ && Objects.equals(authorizedOperations, that.authorizedOperations)
+ && Objects.equals(topologyDescription, that.topologyDescription)
+ && topologyDescriptionStatus == that.topologyDescriptionStatus;
}
@Override
@@ -159,7 +210,9 @@ public class StreamsGroupDescription {
members,
groupState,
coordinator,
- authorizedOperations
+ authorizedOperations,
+ topologyDescription,
+ topologyDescriptionStatus
);
}
@@ -175,6 +228,8 @@ public class StreamsGroupDescription {
", groupState=" + groupState +
", coordinator=" + coordinator +
", authorizedOperations=" +
authorizedOperations.stream().map(AclOperation::toString).collect(Collectors.joining(","))
+
+ ", topologyDescription=" +
topologyDescription.map(Object::toString).orElse("") +
+ ", topologyDescriptionStatus=" + topologyDescriptionStatus +
')';
}
}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescription.java
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescription.java
new file mode 100644
index 00000000000..79d6de44e3a
--- /dev/null
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescription.java
@@ -0,0 +1,422 @@
+/*
+ * 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.HashSet;
+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;
+ // Nodes within a subtopology are unordered, so compare them
set-wise rather than by wire order.
+ return Objects.equals(id, that.id) && Objects.equals(new
HashSet<>(nodes), new HashSet<>(that.nodes));
+ }
+
+ @Override
+ public int hashCode() {
+ // Mirror the order-insensitive equality above; HashSet.hashCode()
is independent of iteration order.
+ return Objects.hash(id, new HashSet<>(nodes));
+ }
+
+ @Override
+ public String toString() {
+ return "Subtopology(id=" + id + ", nodes=" + nodes + ')';
+ }
+ }
+
+ /**
+ * A node of a topology. Can be a {@link Source}, {@link Sink}, or {@link
Processor} node.
+ */
+ public interface Node {
+
+ /**
+ * The name of this node.
+ */
+ String name();
+
+ /**
+ * The names of the successor nodes within the subtopology.
+ */
+ Set<String> successors();
+
+ /**
+ * The names of the predecessor nodes within the subtopology,
reconstructed from the successor relation.
+ */
+ Set<String> predecessors();
+ }
+
+ /**
+ * A source node of a topology.
+ */
+ public static final class Source implements Node {
+
+ private final String name;
+ private final Set<String> topics;
+ private final Set<String> successors;
+ private final Set<String> predecessors;
+
+ public Source(
+ final String name,
+ final Set<String> topics,
+ final Set<String> successors,
+ final Set<String> predecessors
+ ) {
+ this.name = Objects.requireNonNull(name, "name must be non-null");
+ this.topics = Set.copyOf(Objects.requireNonNull(topics, "topics
must be non-null"));
+ this.successors = Set.copyOf(Objects.requireNonNull(successors,
"successors must be non-null"));
+ this.predecessors =
Set.copyOf(Objects.requireNonNull(predecessors, "predecessors must be
non-null"));
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public Set<String> successors() {
+ return successors;
+ }
+
+ @Override
+ public Set<String> predecessors() {
+ return predecessors;
+ }
+
+ /**
+ * The topics this source node reads from. May be empty if the source
topics are dynamically determined.
+ */
+ public Set<String> topics() {
+ return topics;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final Source source = (Source) o;
+ return Objects.equals(name, source.name)
+ && Objects.equals(topics, source.topics)
+ && Objects.equals(successors, source.successors)
+ && Objects.equals(predecessors, source.predecessors);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, topics, successors, predecessors);
+ }
+
+ @Override
+ public String toString() {
+ return "Source(name=" + name + ", topics=" + topics + ",
successors=" + successors + ", predecessors=" + predecessors + ')';
+ }
+ }
+
+ /**
+ * A processor node of a topology.
+ */
+ public static final class Processor implements Node {
+
+ private final String name;
+ private final Set<String> stores;
+ private final Set<String> successors;
+ private final Set<String> predecessors;
+
+ public Processor(
+ final String name,
+ final Set<String> stores,
+ final Set<String> successors,
+ final Set<String> predecessors
+ ) {
+ this.name = Objects.requireNonNull(name, "name must be non-null");
+ this.stores = Set.copyOf(Objects.requireNonNull(stores, "stores
must be non-null"));
+ this.successors = Set.copyOf(Objects.requireNonNull(successors,
"successors must be non-null"));
+ this.predecessors =
Set.copyOf(Objects.requireNonNull(predecessors, "predecessors must be
non-null"));
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public Set<String> successors() {
+ return successors;
+ }
+
+ @Override
+ public Set<String> predecessors() {
+ return predecessors;
+ }
+
+ /**
+ * The names of the state stores accessed by this processor node.
+ */
+ public Set<String> stores() {
+ return stores;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final Processor processor = (Processor) o;
+ return Objects.equals(name, processor.name)
+ && Objects.equals(stores, processor.stores)
+ && Objects.equals(successors, processor.successors)
+ && Objects.equals(predecessors, processor.predecessors);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, stores, successors, predecessors);
+ }
+
+ @Override
+ public String toString() {
+ return "Processor(name=" + name + ", stores=" + stores + ",
successors=" + successors + ", predecessors=" + predecessors + ')';
+ }
+ }
+
+ /**
+ * A sink node of a topology.
+ */
+ public static final class Sink implements Node {
+
+ private final String name;
+ private final Optional<String> topic;
+ private final Set<String> successors;
+ private final Set<String> predecessors;
+
+ public Sink(
+ final String name,
+ final Optional<String> topic,
+ final Set<String> successors,
+ final Set<String> predecessors
+ ) {
+ this.name = Objects.requireNonNull(name, "name must be non-null");
+ this.topic = Objects.requireNonNull(topic, "topic must be
non-null");
+ this.successors = Set.copyOf(Objects.requireNonNull(successors,
"successors must be non-null"));
+ this.predecessors =
Set.copyOf(Objects.requireNonNull(predecessors, "predecessors must be
non-null"));
+ }
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ @Override
+ public Set<String> successors() {
+ return successors;
+ }
+
+ @Override
+ public Set<String> predecessors() {
+ return predecessors;
+ }
+
+ /**
+ * The topic this sink node writes to. Empty if the topic is
dynamically determined.
+ */
+ public Optional<String> topic() {
+ return topic;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final Sink sink = (Sink) o;
+ return Objects.equals(name, sink.name)
+ && Objects.equals(topic, sink.topic)
+ && Objects.equals(successors, sink.successors)
+ && Objects.equals(predecessors, sink.predecessors);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, topic, successors, predecessors);
+ }
+
+ @Override
+ public String toString() {
+ return "Sink(name=" + name + ", topic=" + topic.orElse(null) + ",
successors=" + successors + ", predecessors=" + predecessors + ')';
+ }
+ }
+
+ /**
+ * A global state store, made up of a {@link Source} node and the {@link
Processor} node that maintains the store.
+ */
+ public static class GlobalStore {
+
+ private final Source source;
+ private final Processor processor;
+
+ public GlobalStore(final Source source, final Processor processor) {
+ this.source = Objects.requireNonNull(source, "source must be
non-null");
+ this.processor = Objects.requireNonNull(processor, "processor must
be non-null");
+ }
+
+ /**
+ * The source node providing data to the global store.
+ */
+ public Source source() {
+ return source;
+ }
+
+ /**
+ * The processor node that maintains the global store.
+ */
+ public Processor processor() {
+ return processor;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final GlobalStore that = (GlobalStore) o;
+ return Objects.equals(source, that.source) &&
Objects.equals(processor, that.processor);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(source, processor);
+ }
+
+ @Override
+ public String toString() {
+ return "GlobalStore(source=" + source + ", processor=" + processor
+ ')';
+ }
+ }
+}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescriptionStatus.java
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescriptionStatus.java
new file mode 100644
index 00000000000..22fcf4604a0
--- /dev/null
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/StreamsGroupTopologyDescriptionStatus.java
@@ -0,0 +1,82 @@
+/*
+ * 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;
+
+/**
+ * The status of the topology description that accompanies a {@link
StreamsGroupDescription}.
+ * <p>
+ * The status indicates whether a topology description was requested and, if
so, whether one could
+ * be retrieved. It is paired with {@link
StreamsGroupDescription#topologyDescription()}: the
+ * description is present if and only if the status is {@link #AVAILABLE}.
+ */
[email protected]
+public enum StreamsGroupTopologyDescriptionStatus {
+
+ /**
+ * The topology description was not requested (the caller did not set
+ * {@link
DescribeStreamsGroupsOptions#includeTopologyDescription(boolean)}).
+ */
+ NOT_REQUESTED((byte) 0),
+
+ /**
+ * No topology description is recorded for this group, for example because
no topology description
+ * plugin is configured on the broker or the client has not pushed a
description yet.
+ */
+ NOT_STORED((byte) 1),
+
+ /**
+ * The broker failed to fetch the topology description. See the broker
logs for details.
+ */
+ ERROR((byte) 2),
+
+ /**
+ * The topology description is available and carried in {@link
StreamsGroupDescription#topologyDescription()}.
+ */
+ AVAILABLE((byte) 3);
+
+ private final byte id;
+
+ StreamsGroupTopologyDescriptionStatus(final byte id) {
+ this.id = id;
+ }
+
+ /**
+ * The wire identifier of this status.
+ */
+ public byte id() {
+ return id;
+ }
+
+ /**
+ * Returns the status corresponding to the given wire identifier.
+ *
+ * @param id the wire identifier.
+ * @return the matching status.
+ * @throws IllegalArgumentException if the identifier is unknown.
+ */
+ public static StreamsGroupTopologyDescriptionStatus forId(final byte id) {
+ for (final StreamsGroupTopologyDescriptionStatus status : values()) {
+ if (status.id == id) {
+ return status;
+ }
+ }
+ throw new IllegalArgumentException("Unknown topology description
status id: " + id);
+ }
+}
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
index 69314e6c30f..11d734e5e39 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandler.java
@@ -20,6 +20,8 @@ import org.apache.kafka.clients.admin.StreamsGroupDescription;
import org.apache.kafka.clients.admin.StreamsGroupMemberAssignment;
import org.apache.kafka.clients.admin.StreamsGroupMemberDescription;
import org.apache.kafka.clients.admin.StreamsGroupSubtopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.acl.AclOperation;
@@ -48,14 +50,22 @@ import static
org.apache.kafka.clients.admin.internals.AdminUtils.validAclOperat
public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<CoordinatorKey, StreamsGroupDescription> {
+ static final byte NODE_TYPE_SOURCE = 1;
+ static final byte NODE_TYPE_PROCESSOR = 2;
+ static final byte NODE_TYPE_SINK = 3;
+
private final boolean includeAuthorizedOperations;
+ private final boolean includeTopologyDescription;
private final Logger log;
private final AdminApiLookupStrategy<CoordinatorKey> lookupStrategy;
+
public DescribeStreamsGroupsHandler(
boolean includeAuthorizedOperations,
+ boolean includeTopologyDescription,
LogContext logContext) {
this.includeAuthorizedOperations = includeAuthorizedOperations;
+ this.includeTopologyDescription = includeTopologyDescription;
this.log = logContext.logger(DescribeStreamsGroupsHandler.class);
this.lookupStrategy = new CoordinatorStrategy(CoordinatorType.GROUP,
logContext);
}
@@ -91,7 +101,8 @@ public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<Coordi
}).collect(Collectors.toList());
StreamsGroupDescribeRequestData data = new
StreamsGroupDescribeRequestData()
.setGroupIds(groupIds)
- .setIncludeAuthorizedOperations(includeAuthorizedOperations);
+ .setIncludeAuthorizedOperations(includeAuthorizedOperations)
+ .setIncludeTopologyDescription(includeTopologyDescription);
return new StreamsGroupDescribeRequest.Builder(data);
}
@@ -120,6 +131,28 @@ public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<Coordi
final Set<AclOperation> authorizedOperations =
validAclOperations(describedGroup.authorizedOperations());
+ final StreamsGroupTopologyDescriptionStatus
topologyDescriptionStatus;
+ try {
+ topologyDescriptionStatus =
+
StreamsGroupTopologyDescriptionStatus.forId(describedGroup.topologyDescriptionStatus());
+ } catch (IllegalArgumentException e) {
+ log.error("`DescribeStreamsGroups` response for group id {}
contains an unknown topology description status {}",
+ groupIdKey.idValue,
describedGroup.topologyDescriptionStatus());
+ failed.put(groupIdKey, new IllegalStateException(
+ "Unknown topology description status " +
describedGroup.topologyDescriptionStatus(), e));
+ continue;
+ }
+ final Optional<StreamsGroupTopologyDescription>
topologyDescription;
+ try {
+ topologyDescription =
+ convertTopologyDescription(topologyDescriptionStatus,
describedGroup.topologyDescription());
+ } catch (IllegalStateException e) {
+ log.error("`DescribeStreamsGroups` response for group id {}
contains a topology description that could not be parsed",
+ groupIdKey.idValue, e);
+ failed.put(groupIdKey, e);
+ continue;
+ }
+
final StreamsGroupDescription streamsGroupDescription = new
StreamsGroupDescription(
describedGroup.groupId(),
describedGroup.groupEpoch(),
@@ -129,7 +162,9 @@ public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<Coordi
convertMembers(describedGroup.members()),
GroupState.parse(describedGroup.groupState()),
coordinator,
- authorizedOperations
+ authorizedOperations,
+ topologyDescription,
+ topologyDescriptionStatus
);
completed.put(groupIdKey, streamsGroupDescription);
}
@@ -189,6 +224,86 @@ public class DescribeStreamsGroupsHandler extends
AdminApiHandler.Batched<Coordi
));
}
+ private Optional<StreamsGroupTopologyDescription>
convertTopologyDescription(
+ final StreamsGroupTopologyDescriptionStatus status,
+ final StreamsGroupDescribeResponseData.TopologyDescription
topologyDescription) {
+
+ if (status != StreamsGroupTopologyDescriptionStatus.AVAILABLE) {
+ return Optional.empty();
+ }
+ if (topologyDescription == null) {
+ throw new IllegalStateException("Topology description is missing
despite status AVAILABLE");
+ }
+ final List<StreamsGroupTopologyDescription.Subtopology> subtopologies
= topologyDescription.subtopologies().stream()
+ .map(this::convertTopologySubtopology)
+ .collect(Collectors.toList());
+ final List<StreamsGroupTopologyDescription.GlobalStore> globalStores =
topologyDescription.globalStores().stream()
+ .map(this::convertGlobalStore)
+ .collect(Collectors.toList());
+ return Optional.of(new StreamsGroupTopologyDescription(subtopologies,
globalStores));
+ }
+
+ private StreamsGroupTopologyDescription.Subtopology
convertTopologySubtopology(
+ final
StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology subtopology) {
+ final Map<String, Set<String>> predecessors =
reconstructPredecessors(subtopology.nodes());
+ final List<StreamsGroupTopologyDescription.Node> nodes =
subtopology.nodes().stream()
+ .map(node -> convertTopologyNode(node, predecessors))
+ .collect(Collectors.toList());
+ return new
StreamsGroupTopologyDescription.Subtopology(subtopology.subtopologyId(), nodes);
+ }
+
+ private StreamsGroupTopologyDescription.GlobalStore convertGlobalStore(
+ final
StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore globalStore) {
+ final List<StreamsGroupDescribeResponseData.TopologyDescriptionNode>
pair =
+ List.of(globalStore.source(), globalStore.processor());
+ final Map<String, Set<String>> predecessors =
reconstructPredecessors(pair);
+ final StreamsGroupTopologyDescription.Node source =
convertTopologyNode(globalStore.source(), predecessors);
+ final StreamsGroupTopologyDescription.Node processor =
convertTopologyNode(globalStore.processor(), predecessors);
+ if (!(source instanceof StreamsGroupTopologyDescription.Source)
+ || !(processor instanceof
StreamsGroupTopologyDescription.Processor)) {
+ throw new IllegalStateException("Global store must be composed of
a source and a processor node.");
+ }
+ return new StreamsGroupTopologyDescription.GlobalStore(
+ (StreamsGroupTopologyDescription.Source) source,
+ (StreamsGroupTopologyDescription.Processor) processor
+ );
+ }
+
+ /**
+ * Reconstructs the predecessor relation from the successor lists carried
on the wire. For every node, each of its
+ * successors gains this node as a predecessor.
+ */
+ private Map<String, Set<String>> reconstructPredecessors(
+ final
List<StreamsGroupDescribeResponseData.TopologyDescriptionNode> nodes) {
+ final Map<String, Set<String>> predecessors = new HashMap<>();
+ for (final StreamsGroupDescribeResponseData.TopologyDescriptionNode
node : nodes) {
+ for (final String successor : node.successors()) {
+ predecessors.computeIfAbsent(successor, ignored -> new
HashSet<>()).add(node.name());
+ }
+ }
+ return predecessors;
+ }
+
+ private StreamsGroupTopologyDescription.Node convertTopologyNode(
+ final StreamsGroupDescribeResponseData.TopologyDescriptionNode
node,
+ final Map<String, Set<String>> predecessors) {
+ final Set<String> successors = Set.copyOf(node.successors());
+ final Set<String> nodePredecessors =
predecessors.getOrDefault(node.name(), Set.of());
+ switch (node.nodeType()) {
+ case NODE_TYPE_SOURCE:
+ return new StreamsGroupTopologyDescription.Source(
+ node.name(), Set.copyOf(node.sourceTopics()), successors,
nodePredecessors);
+ case NODE_TYPE_PROCESSOR:
+ return new StreamsGroupTopologyDescription.Processor(
+ node.name(), Set.copyOf(node.stores()), successors,
nodePredecessors);
+ case NODE_TYPE_SINK:
+ return new StreamsGroupTopologyDescription.Sink(
+ node.name(), Optional.ofNullable(node.sinkTopic()),
successors, nodePredecessors);
+ default:
+ throw new IllegalStateException("Unknown topology node type: "
+ node.nodeType());
+ }
+ }
+
private StreamsGroupMemberAssignment.TaskIds convertTaskIds(final
StreamsGroupDescribeResponseData.TaskIds taskIds) {
return new StreamsGroupMemberAssignment.TaskIds(
taskIds.subtopologyId(),
diff --git
a/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
new file mode 100644
index 00000000000..a9df351c87f
--- /dev/null
+++
b/clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClientTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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.GroupState;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class MockAdminClientTest {
+
+ @Test
+ public void testDescribeStreamsGroupsReturnsRegisteredDescription() throws
Exception {
+ String groupId = "stream-group";
+ StreamsGroupDescription description =
newStreamsGroupDescription(groupId);
+
+ try (MockAdminClient admin = new MockAdminClient()) {
+ admin.addStreamsGroupDescription(description);
+
+ StreamsGroupDescription result = admin
+ .describeStreamsGroups(List.of(groupId))
+ .all()
+ .get()
+ .get(groupId);
+
+ assertEquals(description, result);
+ }
+ }
+
+ @Test
+ public void
testDescribeStreamsGroupsReturnsTopologyDescriptionWhenRequested() throws
Exception {
+ String groupId = "stream-group";
+ StreamsGroupDescription description =
newStreamsGroupDescriptionWithTopology(groupId);
+
+ try (MockAdminClient admin = new MockAdminClient()) {
+ admin.addStreamsGroupDescription(description);
+
+ StreamsGroupDescription result = admin
+ .describeStreamsGroups(List.of(groupId), new
DescribeStreamsGroupsOptions().includeTopologyDescription(true))
+ .all()
+ .get()
+ .get(groupId);
+
+ assertEquals(StreamsGroupTopologyDescriptionStatus.AVAILABLE,
result.topologyDescriptionStatus());
+ assertEquals(description.topologyDescription(),
result.topologyDescription());
+ }
+ }
+
+ @Test
+ public void
testDescribeStreamsGroupsOmitsTopologyDescriptionWhenNotRequested() throws
Exception {
+ String groupId = "stream-group";
+ StreamsGroupDescription description =
newStreamsGroupDescriptionWithTopology(groupId);
+
+ try (MockAdminClient admin = new MockAdminClient()) {
+ admin.addStreamsGroupDescription(description);
+
+ StreamsGroupDescription result = admin
+ .describeStreamsGroups(List.of(groupId))
+ .all()
+ .get()
+ .get(groupId);
+
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
result.topologyDescriptionStatus());
+ assertEquals(Optional.empty(), result.topologyDescription());
+ }
+ }
+
+ @Test
+ public void testDescribeStreamsGroupsUnknownGroupFails() {
+ try (MockAdminClient admin = new MockAdminClient()) {
+ ExecutionException exception = assertThrows(
+ ExecutionException.class,
+ () ->
admin.describeStreamsGroups(List.of("missing-group")).all().get());
+ assertInstanceOf(GroupIdNotFoundException.class,
exception.getCause());
+ }
+ }
+
+ private StreamsGroupDescription newStreamsGroupDescription(String groupId)
{
+ return new StreamsGroupDescription(
+ groupId,
+ 0,
+ 0,
+ 0,
+ List.of(),
+ List.of(),
+ GroupState.STABLE,
+ new Node(0, "host", 0),
+ Set.of(),
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED);
+ }
+
+ private StreamsGroupDescription
newStreamsGroupDescriptionWithTopology(String groupId) {
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(List.of(), List.of());
+ return new StreamsGroupDescription(
+ groupId,
+ 0,
+ 0,
+ 0,
+ List.of(),
+ List.of(),
+ GroupState.STABLE,
+ new Node(0, "host", 0),
+ Set.of(),
+ Optional.of(topology),
+ StreamsGroupTopologyDescriptionStatus.AVAILABLE);
+ }
+}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandlerTest.java
b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandlerTest.java
new file mode 100644
index 00000000000..d83a5381c4d
--- /dev/null
+++
b/clients/src/test/java/org/apache/kafka/clients/admin/internals/DescribeStreamsGroupsHandlerTest.java
@@ -0,0 +1,311 @@
+/*
+ * 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.internals;
+
+import org.apache.kafka.clients.admin.StreamsGroupDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.message.StreamsGroupDescribeRequestData;
+import org.apache.kafka.common.message.StreamsGroupDescribeResponseData;
+import org.apache.kafka.common.requests.StreamsGroupDescribeRequest;
+import org.apache.kafka.common.requests.StreamsGroupDescribeResponse;
+import org.apache.kafka.common.utils.internals.LogContext;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class DescribeStreamsGroupsHandlerTest {
+
+ private final LogContext logContext = new LogContext();
+ private final String groupId = "group-id";
+ private final Node coordinator = new Node(1, "host", 1234);
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testBuildRequestSetsIncludeTopologyDescription(boolean
includeTopologyDescription) {
+ DescribeStreamsGroupsHandler handler =
+ new DescribeStreamsGroupsHandler(false,
includeTopologyDescription, logContext);
+
+ StreamsGroupDescribeRequest.Builder builder =
+ handler.buildBatchedRequest(1,
Set.of(CoordinatorKey.byGroupId(groupId)));
+ StreamsGroupDescribeRequestData data = builder.build().data();
+
+ assertEquals(includeTopologyDescription,
data.includeTopologyDescription());
+ assertEquals(List.of(groupId), data.groupIds());
+ }
+
+ @ParameterizedTest
+ @EnumSource(value = StreamsGroupTopologyDescriptionStatus.class, names =
{"NOT_REQUESTED", "NOT_STORED", "ERROR"})
+ public void
testTopologyDescriptionAbsentWhenStatusNotAvailable(StreamsGroupTopologyDescriptionStatus
status) {
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+ .setTopologyDescriptionStatus(status.id())
+ .setTopologyDescription(null);
+
+ StreamsGroupDescription description = describe(describedGroup);
+
+ assertEquals(status, description.topologyDescriptionStatus());
+ assertTrue(description.topologyDescription().isEmpty());
+ }
+
+ @Test
+ public void testTopologyDescriptionParsedWhenAvailable() {
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode source =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("source")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_SOURCE)
+ .setSourceTopics(List.of("input"))
+ .setSuccessors(List.of("processor"));
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode processor =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("processor")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_PROCESSOR)
+ .setStores(List.of("store"))
+ .setSuccessors(List.of("sink"));
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode sink =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("sink")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_SINK)
+ .setSinkTopic("output")
+ .setSuccessors(List.of());
+
+ StreamsGroupDescribeResponseData.TopologyDescription wire =
+ new StreamsGroupDescribeResponseData.TopologyDescription()
+ .setSubtopologies(List.of(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("0")
+ .setNodes(List.of(source, processor, sink))))
+ .setGlobalStores(List.of());
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+
.setTopologyDescriptionStatus(StreamsGroupTopologyDescriptionStatus.AVAILABLE.id())
+ .setTopologyDescription(wire);
+
+ StreamsGroupDescription description = describe(describedGroup);
+
+ assertEquals(StreamsGroupTopologyDescriptionStatus.AVAILABLE,
description.topologyDescriptionStatus());
+ assertTrue(description.topologyDescription().isPresent());
+
+ StreamsGroupTopologyDescription topology =
description.topologyDescription().get();
+ assertEquals(1, topology.subtopologies().size());
+ assertTrue(topology.globalStores().isEmpty());
+
+ StreamsGroupTopologyDescription.Subtopology subtopology =
topology.subtopologies().iterator().next();
+ assertEquals("0", subtopology.id());
+ List<StreamsGroupTopologyDescription.Node> nodes =
List.copyOf(subtopology.nodes());
+ assertEquals(3, nodes.size());
+
+ StreamsGroupTopologyDescription.Source parsedSource =
+ assertInstanceOf(StreamsGroupTopologyDescription.Source.class,
nodes.get(0));
+ assertEquals("source", parsedSource.name());
+ assertEquals(Set.of("input"), parsedSource.topics());
+ assertEquals(Set.of("processor"), parsedSource.successors());
+ // Sources have no predecessors.
+ assertTrue(parsedSource.predecessors().isEmpty());
+
+ StreamsGroupTopologyDescription.Processor parsedProcessor =
+ assertInstanceOf(StreamsGroupTopologyDescription.Processor.class,
nodes.get(1));
+ assertEquals(Set.of("store"), parsedProcessor.stores());
+ assertEquals(Set.of("sink"), parsedProcessor.successors());
+ // Predecessor reconstructed from the source's successor list.
+ assertEquals(Set.of("source"), parsedProcessor.predecessors());
+
+ StreamsGroupTopologyDescription.Sink parsedSink =
+ assertInstanceOf(StreamsGroupTopologyDescription.Sink.class,
nodes.get(2));
+ assertEquals("output", parsedSink.topic().orElse(null));
+ assertTrue(parsedSink.successors().isEmpty());
+ assertEquals(Set.of("processor"), parsedSink.predecessors());
+ }
+
+ @Test
+ public void testGlobalStorePredecessorReconstruction() {
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode globalSource =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("global-source")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_SOURCE)
+ .setSourceTopics(List.of("global-topic"))
+ .setSuccessors(List.of("global-processor"));
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode
globalProcessor =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("global-processor")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_PROCESSOR)
+ .setStores(List.of("global-store"))
+ .setSuccessors(List.of());
+
+ StreamsGroupDescribeResponseData.TopologyDescription wire =
+ new StreamsGroupDescribeResponseData.TopologyDescription()
+ .setSubtopologies(List.of())
+ .setGlobalStores(List.of(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore()
+ .setSource(globalSource)
+ .setProcessor(globalProcessor)));
+
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+
.setTopologyDescriptionStatus(StreamsGroupTopologyDescriptionStatus.AVAILABLE.id())
+ .setTopologyDescription(wire);
+
+ StreamsGroupTopologyDescription topology =
describe(describedGroup).topologyDescription().orElseThrow();
+ assertEquals(1, topology.globalStores().size());
+
+ StreamsGroupTopologyDescription.GlobalStore globalStore =
topology.globalStores().iterator().next();
+ assertTrue(globalStore.source().predecessors().isEmpty());
+ assertEquals(Set.of("global-processor"),
globalStore.source().successors());
+ assertEquals(Set.of("global-source"),
globalStore.processor().predecessors());
+ }
+
+ @Test
+ public void testTopologyDescriptionNotRequestedByDefault() {
+ StreamsGroupDescription description = describe(newDescribedGroup());
+ assertEquals(StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED,
description.topologyDescriptionStatus());
+ assertFalse(description.topologyDescription().isPresent());
+ }
+
+ @Test
+ public void testUnknownTopologyDescriptionStatusFailsOnlyAffectedGroup() {
+ // A future broker version may add a topology description status id
that this client does not know about.
+ byte unknownStatusId = (byte) 99;
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+ .setTopologyDescriptionStatus(unknownStatusId)
+ .setTopologyDescription(null);
+
+ DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(false, true, logContext);
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ AdminApiHandler.ApiResult<CoordinatorKey, StreamsGroupDescription>
result = handler.handleResponse(
+ coordinator,
+ Set.of(key),
+ new StreamsGroupDescribeResponse(new
StreamsGroupDescribeResponseData()
+ .setGroups(List.of(describedGroup))));
+
+ // The unknown status must surface as a per-group failure rather than
propagating an exception.
+ assertTrue(result.completedKeys.isEmpty());
+ assertEquals(Set.of(key), result.failedKeys.keySet());
+ assertInstanceOf(IllegalStateException.class,
result.failedKeys.get(key));
+ }
+
+ @Test
+ public void
testAvailableStatusWithMissingTopologyDescriptionFailsOnlyAffectedGroup() {
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+
.setTopologyDescriptionStatus(StreamsGroupTopologyDescriptionStatus.AVAILABLE.id())
+ .setTopologyDescription(null);
+
+ DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(false, true, logContext);
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ AdminApiHandler.ApiResult<CoordinatorKey, StreamsGroupDescription>
result = handler.handleResponse(
+ coordinator,
+ Set.of(key),
+ new StreamsGroupDescribeResponse(new
StreamsGroupDescribeResponseData()
+ .setGroups(List.of(describedGroup))));
+
+ assertTrue(result.completedKeys.isEmpty());
+ assertEquals(Set.of(key), result.failedKeys.keySet());
+ assertInstanceOf(IllegalStateException.class,
result.failedKeys.get(key));
+ }
+
+ @Test
+ public void testUnknownTopologyNodeTypeFailsOnlyAffectedGroup() {
+ // A future broker version may add a topology node type that this
client does not know about.
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode unknownNode =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("mystery")
+ .setNodeType((byte) 99)
+ .setSuccessors(List.of());
+ StreamsGroupDescribeResponseData.TopologyDescription wire =
+ new StreamsGroupDescribeResponseData.TopologyDescription()
+ .setSubtopologies(List.of(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionSubtopology()
+ .setSubtopologyId("0")
+ .setNodes(List.of(unknownNode))))
+ .setGlobalStores(List.of());
+
+ assertGroupFailed(wire);
+ }
+
+ @Test
+ public void testMalformedGlobalStoreFailsOnlyAffectedGroup() {
+ // A global store must be a source/processor pair; here the "source"
is itself a sink node.
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode notASource =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("global-sink")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_SINK)
+ .setSinkTopic("output")
+ .setSuccessors(List.of());
+ StreamsGroupDescribeResponseData.TopologyDescriptionNode processor =
+ new StreamsGroupDescribeResponseData.TopologyDescriptionNode()
+ .setName("global-processor")
+ .setNodeType(DescribeStreamsGroupsHandler.NODE_TYPE_PROCESSOR)
+ .setStores(List.of("global-store"))
+ .setSuccessors(List.of());
+ StreamsGroupDescribeResponseData.TopologyDescription wire =
+ new StreamsGroupDescribeResponseData.TopologyDescription()
+ .setSubtopologies(List.of())
+ .setGlobalStores(List.of(
+ new
StreamsGroupDescribeResponseData.TopologyDescriptionGlobalStore()
+ .setSource(notASource)
+ .setProcessor(processor)));
+
+ assertGroupFailed(wire);
+ }
+
+ private void
assertGroupFailed(StreamsGroupDescribeResponseData.TopologyDescription wire) {
+ StreamsGroupDescribeResponseData.DescribedGroup describedGroup =
newDescribedGroup()
+
.setTopologyDescriptionStatus(StreamsGroupTopologyDescriptionStatus.AVAILABLE.id())
+ .setTopologyDescription(wire);
+
+ DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(false, true, logContext);
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ AdminApiHandler.ApiResult<CoordinatorKey, StreamsGroupDescription>
result = handler.handleResponse(
+ coordinator,
+ Set.of(key),
+ new StreamsGroupDescribeResponse(new
StreamsGroupDescribeResponseData()
+ .setGroups(List.of(describedGroup))));
+
+ assertTrue(result.completedKeys.isEmpty());
+ assertEquals(Set.of(key), result.failedKeys.keySet());
+ assertInstanceOf(IllegalStateException.class,
result.failedKeys.get(key));
+ }
+
+ private StreamsGroupDescribeResponseData.DescribedGroup
newDescribedGroup() {
+ return new StreamsGroupDescribeResponseData.DescribedGroup()
+ .setGroupId(groupId)
+ .setGroupState("Stable")
+ .setGroupEpoch(1)
+ .setAssignmentEpoch(1)
+ .setTopology(new
StreamsGroupDescribeResponseData.Topology().setEpoch(1).setSubtopologies(List.of()));
+ }
+
+ private StreamsGroupDescription
describe(StreamsGroupDescribeResponseData.DescribedGroup describedGroup) {
+ DescribeStreamsGroupsHandler handler = new
DescribeStreamsGroupsHandler(false, true, logContext);
+ CoordinatorKey key = CoordinatorKey.byGroupId(groupId);
+ AdminApiHandler.ApiResult<CoordinatorKey, StreamsGroupDescription>
result = handler.handleResponse(
+ coordinator,
+ Set.of(key),
+ new StreamsGroupDescribeResponse(new
StreamsGroupDescribeResponseData()
+ .setGroups(List.of(describedGroup))));
+ assertTrue(result.failedKeys.isEmpty(), () -> "Unexpected failures: "
+ result.failedKeys);
+ return result.completedKeys.get(key);
+ }
+}
diff --git
a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
index bb460855df0..f2f6e4a769b 100644
---
a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
+++
b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java
@@ -39,6 +39,7 @@ import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.DelegationTokenNotFoundException;
+import org.apache.kafka.common.errors.GroupIdNotFoundException;
import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
import org.apache.kafka.common.errors.InvalidReplicationFactorException;
import org.apache.kafka.common.errors.InvalidRequestException;
@@ -100,6 +101,7 @@ public class MockAdminClient extends AdminClient {
private final Map<String, Map<String, String>> groupConfigs;
private final Map<String, String> defaultGroupConfigs;
private final List<KafkaMetric> addedMetrics = new ArrayList<>();
+ private final Map<String, StreamsGroupDescription>
streamsGroupDescriptions = new HashMap<>();
private Node controller;
private int timeoutNextRequests = 0;
@@ -1470,11 +1472,48 @@ public class MockAdminClient extends AdminClient {
throw new UnsupportedOperationException("Not implemented yet");
}
+ /**
+ * Registers a {@link StreamsGroupDescription} to be returned by {@link
#describeStreamsGroups}.
+ */
+ public synchronized void
addStreamsGroupDescription(StreamsGroupDescription description) {
+ streamsGroupDescriptions.put(description.groupId(), description);
+ }
+
@Override
public synchronized DescribeStreamsGroupsResult
describeStreamsGroups(Collection<String> groupIds, DescribeStreamsGroupsOptions
options) {
- throw new UnsupportedOperationException("Not implemented yet");
+ Map<String, KafkaFuture<StreamsGroupDescription>> futures = new
HashMap<>();
+ for (String groupId : groupIds) {
+ KafkaFutureImpl<StreamsGroupDescription> future = new
KafkaFutureImpl<>();
+ StreamsGroupDescription description =
streamsGroupDescriptions.get(groupId);
+ if (description != null) {
+ if (!options.includeTopologyDescription()) {
+ description = withoutTopologyDescription(description);
+ }
+ future.complete(description);
+ } else {
+ future.completeExceptionally(new
GroupIdNotFoundException("Group " + groupId + " not found."));
+ }
+ futures.put(groupId, future);
+ }
+ return new DescribeStreamsGroupsResult(futures);
}
-
+
+ private static StreamsGroupDescription
withoutTopologyDescription(StreamsGroupDescription description) {
+ return new StreamsGroupDescription(
+ description.groupId(),
+ description.groupEpoch(),
+ description.targetAssignmentEpoch(),
+ description.topologyEpoch(),
+ description.subtopologies(),
+ description.members(),
+ description.groupState(),
+ description.coordinator(),
+ description.authorizedOperations(),
+ Optional.empty(),
+ StreamsGroupTopologyDescriptionStatus.NOT_REQUESTED
+ );
+ }
+
@Override
public synchronized DescribeClassicGroupsResult
describeClassicGroups(Collection<String> groupIds, DescribeClassicGroupsOptions
options) {
throw new UnsupportedOperationException("Not implemented yet");
diff --git a/docs/streams/developer-guide/kafka-streams-group-sh.md
b/docs/streams/developer-guide/kafka-streams-group-sh.md
index e11f6b178b4..fc678c889a4 100644
--- a/docs/streams/developer-guide/kafka-streams-group-sh.md
+++ b/docs/streams/developer-guide/kafka-streams-group-sh.md
@@ -41,6 +41,7 @@ A **Streams group** is a broker‑coordinated group type for
Kafka Streams that
* Group state, group epoch, target assignment epoch (with `--state`,
`--verbose` for additional details).
* Per‑member info such as epochs, current vs target assignments, and
whether a member still uses the classic protocol (with `--members` and
`--verbose`).
* Input‑topic offsets and lag (with `--offsets`), to understand how far
behind processing is.
+ * The processing topology, as recorded by the broker's topology
description plugin (with `--topology`), in a format that mirrors
`Topology#describe()`.
* **Reset input‑topic offsets** for a Streams group to control reprocessing
boundaries using precise specifiers (earliest, latest, to‑offset, to‑datetime,
by‑duration, shift‑by, from‑file). Requires `--dry-run` or `--execute` and
inactive instances.
* **Delete offsets** for input topics to force re‑consumption on next start.
* **Delete a Streams group** to clean up broker‑side Streams metadata
(offsets, topology, assignments). Optionally delete all, or a subset of,
**internal topics** at the same time using `--internal-topics`.
@@ -83,6 +84,10 @@ Inspecting group's state, members, and lag
# Describe a group: input-topic offsets and lag
kafka-streams-groups.sh --bootstrap-server localhost:9092 \
--describe --group my-streams-app --offsets
+
+ # Describe a group: processing topology
+ kafka-streams-groups.sh --bootstrap-server localhost:9092 \
+ --describe --group my-streams-app --topology
## Reset input-topic offsets (preview, then apply) {#reset-offsets}
@@ -140,7 +145,7 @@ Delete broker-side Streams metadata for a group and
optionally remove a subset o
* `--list`: List Streams groups. Use `--state` to display/filter by state.
* `--describe`: Describe a group selected by `--group`. Combine with:
- * `--state` (group state and epochs), `--members` (members and
assignments), `--offsets` (input and repartition topics offsets/lag).
+ * `--state` (group state and epochs), `--members` (members and
assignments), `--offsets` (input and repartition topics offsets/lag),
`--topology` (processing topology recorded by the broker's topology description
plugin).
* `--verbose` for additional details (e.g., leader epochs where
applicable).
* `--reset-offsets`: Reset input-topic offsets (one group at a time;
instances should be inactive). Choose exactly one specifier:
* `--to-earliest`, `--to-latest`, `--to-current`, `--to-offset <n>`
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 92d03758fd0..083e71e3019 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
@@ -109,7 +109,7 @@ public class StreamsGroupCommand {
if (numberOfActions != 1)
throw new IllegalArgumentException("Command must include
exactly one action: --list, --describe, --delete, --reset-offsets, or
--delete-offsets.");
- run(opts);
+ exitCode = run(opts);
} catch (IllegalArgumentException | OptionException e) {
System.err.println(e.getMessage());
if (opts != null) {
@@ -128,12 +128,12 @@ public class StreamsGroupCommand {
return exitCode;
}
- public static void run(StreamsGroupCommandOptions opts) throws
ExecutionException, InterruptedException {
+ public static int run(StreamsGroupCommandOptions opts) throws
ExecutionException, InterruptedException {
try (StreamsGroupService streamsGroupService = new
StreamsGroupService(opts, Map.of())) {
if (opts.options.has(opts.listOpt)) {
streamsGroupService.listGroups();
} else if (opts.options.has(opts.describeOpt)) {
- streamsGroupService.describeGroups();
+ return streamsGroupService.describeGroups();
} else if (opts.options.has(opts.resetOffsetsOpt)) {
Map<String, Map<TopicPartition, OffsetAndMetadata>>
offsetsToReset = streamsGroupService.resetOffsets();
if (opts.options.has(opts.exportOpt)) {
@@ -149,6 +149,7 @@ public class StreamsGroupCommand {
throw new IllegalArgumentException("Unknown action!");
}
}
+ return 0;
}
static void printOffsetsToReset(Map<String, Map<TopicPartition,
OffsetAndMetadata>> groupAssignmentsToReset) {
@@ -262,15 +263,21 @@ public class StreamsGroupCommand {
}
}
- public void describeGroups() throws ExecutionException,
InterruptedException {
+ public int describeGroups() throws ExecutionException,
InterruptedException {
List<String> groupIds = opts.options.has(opts.allGroupsOpt)
? new ArrayList<>(listStreamsGroups())
: new ArrayList<>(opts.options.valuesOf(opts.groupOpt));
+ int exitCode = 0;
if (!groupIds.isEmpty()) {
+ boolean topology = opts.options.has(opts.topologyOpt);
for (String groupId : groupIds) {
- StreamsGroupDescription description =
getDescribeGroup(groupId);
+ StreamsGroupDescription description =
getDescribeGroup(groupId, topology);
boolean verbose = opts.options.has(opts.verboseOpt);
- if (opts.options.has(opts.membersOpt)) {
+ if (topology) {
+ if (!printTopology(description)) {
+ exitCode = 1;
+ }
+ } else if (opts.options.has(opts.membersOpt)) {
printMembers(description, verbose);
} else if (opts.options.has(opts.stateOpt)) {
printStates(description, verbose);
@@ -279,16 +286,44 @@ public class StreamsGroupCommand {
}
}
}
+ 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()));
+ return true;
+ case NOT_STORED:
+ printError("No topology description is stored for streams
group '" + description.groupId() + "'.", Optional.empty());
+ return false;
+ case ERROR:
+ printError("The broker failed to fetch the topology
description for streams group '" + description.groupId()
+ + "'. See the broker logs for details.",
Optional.empty());
+ return false;
+ default:
+ printError("No topology description is available for
streams group '" + description.groupId()
+ + "' (status: " +
description.topologyDescriptionStatus() + ").", Optional.empty());
+ return false;
+ }
+ }
+
private void printMembers(StreamsGroupDescription description, boolean
verbose) {
final int groupLen = Math.max(15, description.groupId().length());
int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
diff --git
a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommandOptions.java
b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommandOptions.java
index 66546961164..8d2d9fb18cb 100644
---
a/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommandOptions.java
+++
b/tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommandOptions.java
@@ -49,6 +49,8 @@ public class StreamsGroupCommandOptions extends
CommandDefaultOptions {
private static final String MEMBERS_DOC = "Describe members of the group.
This option may be used with the --describe option only.";
private static final String OFFSETS_DOC = "Describe the group and list all
topic partitions in the group along with their offset information." +
"This is the default sub-action and may be used with the --describe
option only.";
+ private static final String TOPOLOGY_DOC = "Describe the topology of the
streams group, as recorded by the broker's topology description plugin. " +
+ "This option may be used with the '--describe' option only.";
private static final String RESET_OFFSETS_DOC = "Reset offsets of streams
group. The instances should be inactive." + NL +
"Has 2 execution options: --dry-run to plan which offsets to reset,
and --execute to update the offsets." + NL +
"If you use --execute, all internal topics linked to the group will
also be deleted." + NL +
@@ -91,6 +93,7 @@ public class StreamsGroupCommandOptions extends
CommandDefaultOptions {
final OptionSpec<String> stateOpt;
final OptionSpec<Void> membersOpt;
final OptionSpec<Void> offsetsOpt;
+ final OptionSpec<Void> topologyOpt;
final OptionSpec<Void> resetOffsetsOpt;
final OptionSpec<Long> resetToOffsetOpt;
final OptionSpec<String> resetFromFileOpt;
@@ -158,6 +161,8 @@ public class StreamsGroupCommandOptions extends
CommandDefaultOptions {
.availableIf(describeOpt);
offsetsOpt = parser.accepts("offsets", OFFSETS_DOC)
.availableIf(describeOpt);
+ topologyOpt = parser.accepts("topology", TOPOLOGY_DOC)
+ .availableIf(describeOpt);
resetOffsetsOpt = parser.accepts("reset-offsets", RESET_OFFSETS_DOC);
resetToOffsetOpt = parser.accepts("to-offset", RESET_TO_OFFSET_DOC)
.withRequiredArg()
@@ -235,7 +240,7 @@ public class StreamsGroupCommandOptions extends
CommandDefaultOptions {
if ((options.has(dryRunOpt) || options.has(executeOpt)) &&
!options.has(resetOffsetsOpt))
CommandLineUtils.printUsageAndExit(parser, "Only Option " +
resetOffsetsOpt + " accepts " + executeOpt + " or " + dryRunOpt);
- CommandLineUtils.checkInvalidArgs(parser, options, listOpt,
membersOpt, offsetsOpt);
+ CommandLineUtils.checkInvalidArgs(parser, options, listOpt,
membersOpt, offsetsOpt, topologyOpt);
CommandLineUtils.checkInvalidArgs(parser, options, groupOpt,
minus(allGroupSelectionScopeOpts, groupOpt));
CommandLineUtils.checkInvalidArgs(parser, options, groupOpt,
minus(allStreamsGroupLevelOpts, describeOpt, deleteOpt, resetOffsetsOpt));
CommandLineUtils.checkInvalidArgs(parser, options, inputTopicOpt,
minus(allStreamsGroupLevelOpts, resetOffsetsOpt));
@@ -247,7 +252,7 @@ public class StreamsGroupCommandOptions extends
CommandDefaultOptions {
if (!options.has(groupOpt) && !options.has(allGroupsOpt))
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes one of these options: " +
allGroupSelectionScopeOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(",
")));
- List<OptionSpec<?>> mutuallyExclusiveOpts = List.of(membersOpt,
offsetsOpt, stateOpt);
+ List<OptionSpec<?>> mutuallyExclusiveOpts = List.of(membersOpt,
offsetsOpt, stateOpt, topologyOpt);
if (mutuallyExclusiveOpts.stream().mapToInt(o -> options.has(o) ? 1 :
0).sum() > 1) {
CommandLineUtils.printUsageAndExit(parser,
"Option " + describeOpt + " takes at most one of these
options: " +
mutuallyExclusiveOpts.stream().map(Object::toString).sorted().collect(Collectors.joining(",
")));
diff --git
a/tools/src/main/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatter.java
b/tools/src/main/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatter.java
new file mode 100644
index 00000000000..970d0a6af23
--- /dev/null
+++
b/tools/src/main/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatter.java
@@ -0,0 +1,108 @@
+/*
+ * 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);
+ }
+ 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');
+ 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) {
+ if (names.isEmpty()) {
+ return "none";
+ }
+ return String.join(", ", sorted(names));
+ }
+}
diff --git
a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
index 20f6ed26e60..55e2dde048e 100644
---
a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
+++
b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java
@@ -37,6 +37,8 @@ import org.apache.kafka.clients.admin.StreamsGroupDescription;
import org.apache.kafka.clients.admin.StreamsGroupMemberAssignment;
import org.apache.kafka.clients.admin.StreamsGroupMemberDescription;
import org.apache.kafka.clients.admin.StreamsGroupSubtopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescription;
+import org.apache.kafka.clients.admin.StreamsGroupTopologyDescriptionStatus;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.GroupState;
@@ -48,8 +50,10 @@ import org.apache.kafka.common.TopicPartitionInfo;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.internals.KafkaFutureImpl;
import org.apache.kafka.test.TestUtils;
+import org.apache.kafka.tools.ToolsTestUtils;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.MockedStatic;
import java.util.ArrayList;
@@ -190,6 +194,64 @@ public class StreamsGroupCommandTest {
service.close();
}
+ @Test
+ public void testDescribeStreamsGroupsWithTopologyAvailable() throws
Exception {
+ String group = "foo-group";
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new StreamsGroupTopologyDescription.Subtopology("0",
List.of(
+ new StreamsGroupTopologyDescription.Source("source",
Set.of("input"), Set.of("sink"), Set.of()),
+ new StreamsGroupTopologyDescription.Sink("sink",
Optional.of("output"), Set.of(), Set.of("source"))))),
+ List.of());
+ StreamsGroupDescription exp = new StreamsGroupDescription(
+ group, 0, 0, 0, List.of(), List.of(), GroupState.STABLE, new
Node(0, "bar", 0), null,
+ Optional.of(topology),
StreamsGroupTopologyDescriptionStatus.AVAILABLE);
+
+ Admin admin = mock(KafkaAdminClient.class);
+ DescribeStreamsGroupsResult result =
mock(DescribeStreamsGroupsResult.class);
+
when(result.all()).thenReturn(KafkaFuture.completedFuture(Map.of(group, exp)));
+ ArgumentCaptor<DescribeStreamsGroupsOptions> optionsCaptor =
ArgumentCaptor.forClass(DescribeStreamsGroupsOptions.class);
+ when(admin.describeStreamsGroups(anyCollection(),
optionsCaptor.capture())).thenReturn(result);
+
+ StreamsGroupCommandOptions opts = new StreamsGroupCommandOptions(
+ new String[]{"--bootstrap-server", BOOTSTRAP_SERVERS, "--group",
group, "--describe", "--topology"});
+ StreamsGroupCommand.StreamsGroupService service = new
StreamsGroupCommand.StreamsGroupService(opts, admin);
+
+ String output = ToolsTestUtils.grabConsoleOutput(() -> {
+ try {
+ assertEquals(0, service.describeGroups());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ assertTrue(optionsCaptor.getValue().includeTopologyDescription(),
"Topology description should be requested.");
+ assertTrue(output.contains("Sub-topology: 0"), "Unexpected output: " +
output);
+ assertTrue(output.contains("Source: source (topics: [input])"),
"Unexpected output: " + output);
+ assertTrue(output.contains("Sink: sink (topic: output)"), "Unexpected
output: " + output);
+ service.close();
+ }
+
+ @Test
+ public void testDescribeStreamsGroupsWithTopologyNotStored() throws
Exception {
+ String group = "foo-group";
+ StreamsGroupDescription exp = new StreamsGroupDescription(
+ group, 0, 0, 0, List.of(), List.of(), GroupState.STABLE, new
Node(0, "bar", 0), null,
+ Optional.empty(),
StreamsGroupTopologyDescriptionStatus.NOT_STORED);
+
+ Admin admin = mock(KafkaAdminClient.class);
+ DescribeStreamsGroupsResult result =
mock(DescribeStreamsGroupsResult.class);
+
when(result.all()).thenReturn(KafkaFuture.completedFuture(Map.of(group, exp)));
+ when(admin.describeStreamsGroups(anyCollection(),
any(DescribeStreamsGroupsOptions.class))).thenReturn(result);
+
+ StreamsGroupCommandOptions opts = new StreamsGroupCommandOptions(
+ new String[]{"--bootstrap-server", BOOTSTRAP_SERVERS, "--group",
group, "--describe", "--topology"});
+ StreamsGroupCommand.StreamsGroupService service = new
StreamsGroupCommand.StreamsGroupService(opts, admin);
+
+ // A missing topology description must surface a non-zero exit code.
+ assertEquals(1, service.describeGroups());
+ service.close();
+ }
+
@Test
public void testDescribeStreamsGroupsGetOffsets() throws Exception {
String groupId = "group1";
diff --git
a/tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java
b/tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java
new file mode 100644
index 00000000000..b6503382c33
--- /dev/null
+++
b/tools/src/test/java/org/apache/kafka/tools/streams/TopologyDescriptionFormatterTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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" +
+ " --> none\n" +
+ " <-- global-source\n" +
+ "\n";
+
+ assertEquals(expected, TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatMultipleGlobalStoresAreSeparatedByBlankLine() {
+ Source source0 = new Source("global-source-0", Set.of("topic-0"),
Set.of("global-processor-0"), Set.of());
+ Processor processor0 = new Processor("global-processor-0",
Set.of("store-0"), Set.of(), Set.of("global-source-0"));
+ Source source1 = new Source("global-source-1", Set.of("topic-1"),
Set.of("global-processor-1"), Set.of());
+ Processor processor1 = new Processor("global-processor-1",
Set.of("store-1"), Set.of(), Set.of("global-source-1"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(),
+ List.of(new GlobalStore(source0, processor0), new
GlobalStore(source1, processor1)));
+
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0 for global store (will not generate tasks)\n" +
+ " Source: global-source-0 (topics: [topic-0])\n" +
+ " --> global-processor-0\n" +
+ " Processor: global-processor-0 (stores: [store-0])\n" +
+ " --> none\n" +
+ " <-- global-source-0\n" +
+ "\n" +
+ " Sub-topology: 1 for global store (will not generate tasks)\n" +
+ " Source: global-source-1 (topics: [topic-1])\n" +
+ " --> global-processor-1\n" +
+ " Processor: global-processor-1 (stores: [store-1])\n" +
+ " --> none\n" +
+ " <-- global-source-1\n" +
+ "\n", 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());
+
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source-0 (topics: [input-0])\n" +
+ " --> none\n" +
+ "\n" +
+ " Sub-topology: 1\n" +
+ " Source: source-1 (topics: [input-1])\n" +
+ " --> none\n" +
+ "\n", TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatRendersEmptySuccessorsAndPredecessorsAsNone() {
+ Processor processor = new Processor("processor", Set.of("store"),
Set.of(), Set.of());
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new Subtopology("0", List.<Node>of(processor))),
+ List.of());
+
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Processor: processor (stores: [store])\n" +
+ " --> none\n" +
+ " <-- none\n" +
+ "\n", TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatEmptyTopologyHasNoTrailingSpace() {
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(List.of(), List.of());
+ assertEquals("Topologies:\n",
TopologyDescriptionFormatter.format(topology));
+ }
+
+ @Test
+ public void testFormatSinkWithoutTopicAndMultipleSuccessors() {
+ Source source = new Source("source", Set.of("t2", "t1"), Set.of("p2",
"p1"), Set.of());
+ Sink sink = new Sink("sink", Optional.empty(), Set.of(),
Set.of("source"));
+ StreamsGroupTopologyDescription topology = new
StreamsGroupTopologyDescription(
+ List.of(new Subtopology("0", List.<Node>of(source, sink))),
+ List.of());
+
+ String formatted = TopologyDescriptionFormatter.format(topology);
+
+ assertEquals("Topologies:\n" +
+ " Sub-topology: 0\n" +
+ " Source: source (topics: [t1, t2])\n" +
+ " --> p1, p2\n" +
+ " Sink: sink (topic: null)\n" +
+ " <-- source\n" +
+ "\n", formatted);
+ }
+}
