This is an automated email from the ASF dual-hosted git repository.
frankvicky 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 90b1ab7fc78 KAFKA-20630: Add documentation for topology description
plugin feature (#22776)
90b1ab7fc78 is described below
commit 90b1ab7fc782b6530d9a71fb2437e85fc3061985
Author: Alieh Saeedi <[email protected]>
AuthorDate: Tue Jul 14 14:08:49 2026 +0200
KAFKA-20630: Add documentation for topology description plugin feature
(#22776)
Documents the streams group topology description plugin
([KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin));
docs subtask of KAFKA-20618.
### Changes
- **New page**
`docs/streams/developer-guide/topology-description-plugin.md`: feature
overview, the push/describe cycle (heartbeat solicitation via
`TopologyDescriptionRequired`, the
`StreamsGroupTopologyDescriptionUpdate` RPC, permanent vs. transient
plugin failures and the per-group back-off, describe via
`StreamsGroupDescribe` v1), broker and client configuration, plugin
implementation guidelines, Admin API and CLI usage,
`TopologyDescriptionStatus` interpretation, `GROUP_DELETION_FAILED`
semantics, and troubleshooting.
- `config-streams.md`: add `topology.description.push.enabled` to the
config table and the detailed entries.
- `kafka-streams-group-sh.md`: document `--topology` output format,
broker requirements (4.4+, plugin configured), and the tool's error
messages / exit-code behavior.
- `streams-rebalance-protocol.md`: list the feature as supported, add
`group.streams.topology.description.plugin.class` to the broker-config
list, and link the new page from the administration section.
- `getting-started/upgrade.md`: notable-changes entry for 4.4.0 covering
the new RPC, the `StreamsGroupDescribe`/`StreamsGroupHeartbeat` v1
bumps, `DeleteGroups` v3 with per-group `ErrorMessage`, the new error
codes `GROUP_DELETION_FAILED` (134) and
`STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` (135), and the new
broker/client configs.
- `streams/upgrade-guide.md`: 4.4.0 entry from the application
perspective.
- `operations/basic-kafka-operations.md`: point streams-group management
at `kafka-streams-groups.sh`, including `--topology`.
- `scala-migration.md`: weight bump only (16 → 17) to make room for the
new page in the developer-guide ordering.
All described behavior was verified against the implementation
(`StreamsGroupTopologyDescriptionPlugin` SPI,
`StreamsGroupTopologyDescriptionManager`, `StreamsGroupCommand`, the
message JSON schemas, and the config definitions).
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Reviewers: TengYao Chi <[email protected]>
---------
Co-authored-by: Claude Fable 5 <[email protected]>
---
docs/getting-started/upgrade.md | 1 +
docs/operations/basic-kafka-operations.md | 2 +-
docs/operations/monitoring.md | 208 +++++++++++++++++++++
docs/streams/developer-guide/config-streams.md | 23 +++
.../developer-guide/kafka-streams-group-sh.md | 23 ++-
docs/streams/developer-guide/scala-migration.md | 2 +-
.../developer-guide/streams-rebalance-protocol.md | 7 +
.../developer-guide/topology-description-plugin.md | 195 +++++++++++++++++++
docs/streams/upgrade-guide.md | 2 +
9 files changed, 460 insertions(+), 3 deletions(-)
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index f2594cdd3e8..bd429ee95b6 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -44,6 +44,7 @@ type: docs
For further details, please refer to
[KIP-1301](https://cwiki.apache.org/confluence/x/Z5U8G).
* The broker-side OAUTHBEARER JWT validator now fails fast at startup when a
JWKS endpoint (`sasl.oauthbearer.jwks.endpoint.url`) is configured but
`sasl.oauthbearer.expected.audience` or `sasl.oauthbearer.expected.issuer` is
not set. Brokers that previously started without these settings will now fail
to start until they are configured. To intentionally accept tokens regardless
of their audience or issuer, set the new
`sasl.oauthbearer.allow.unverified.audience` or `sasl.oauthbearer.a [...]
* When clients connect to the cluster, they now include cluster and node
information to enable detection and handling of misrouted connections. For
further details, please refer to
[KIP-1242](https://cwiki.apache.org/confluence/x/W4LMFw).
+ * Brokers can now record a human-readable description of each streams
group's processing topology via a pluggable backend, retrievable through
`Admin#describeStreamsGroups` and `kafka-streams-groups.sh --describe
--topology`. The feature is disabled unless the new broker configuration
`group.streams.topology.description.plugin.class` is set to a
`StreamsGroupTopologyDescriptionPlugin` implementation; on the client side, the
new Kafka Streams configuration `topology.description.push.ena [...]
## Upgrading to 4.3.0
diff --git a/docs/operations/basic-kafka-operations.md
b/docs/operations/basic-kafka-operations.md
index 4ed37d87417..d5146f7404d 100644
--- a/docs/operations/basic-kafka-operations.md
+++ b/docs/operations/basic-kafka-operations.md
@@ -151,7 +151,7 @@ my-topic 2 2 3
1
## Managing groups
-With the GroupCommand tool, we can list groups of all types, including
consumer groups, share groups and streams groups. Each type of group has its
own tool for administering groups of that type. For example, to list all groups
in the cluster:
+With the GroupCommand tool, we can list groups of all types, including
consumer groups, share groups and streams groups. Each type of group has its
own tool for administering groups of that type. For streams groups, use the
[bin/kafka-streams-groups.sh](/{version}/streams/developer-guide/kafka-streams-group-sh/)
tool to list, describe (including the group's processing topology via
`--describe --topology`, when a [topology description
plugin](/{version}/streams/developer-guide/topology-de [...]
```bash
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
diff --git a/docs/operations/monitoring.md b/docs/operations/monitoring.md
index 3f5fb0ec66f..92badfae27a 100644
--- a/docs/operations/monitoring.md
+++ b/docs/operations/monitoring.md
@@ -1348,6 +1348,214 @@ Total number of Streams Group Rebalances
<tr>
<td>
+Topology Description Set Success Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-success-rate
+</td>
+<td>
+
+The rate of successful setTopology plugin calls (driven by client
topology-description pushes)
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Set Success Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-success-count
+</td>
+<td>
+
+The total number of successful setTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Set Error Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-error-rate
+</td>
+<td>
+
+The rate of failed setTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Set Error Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-error-count
+</td>
+<td>
+
+The total number of failed setTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Get Success Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-success-rate
+</td>
+<td>
+
+The rate of successful getTopology plugin calls (driven by describe requests)
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Get Success Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-success-count
+</td>
+<td>
+
+The total number of successful getTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Get Error Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-error-rate
+</td>
+<td>
+
+The rate of failed getTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Get Error Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-error-count
+</td>
+<td>
+
+The total number of failed getTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Delete Success Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-success-rate
+</td>
+<td>
+
+The rate of successful deleteTopology plugin calls (driven by group deletion
and cleanup)
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Delete Success Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-success-count
+</td>
+<td>
+
+The total number of successful deleteTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Delete Error Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-error-rate
+</td>
+<td>
+
+The rate of failed deleteTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Delete Error Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-error-count
+</td>
+<td>
+
+The total number of failed deleteTopology plugin calls
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Cleanup Cycle Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-cycle-rate
+</td>
+<td>
+
+The rate of periodic topology-description cleanup cycles run by the coordinator
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Cleanup Cycle Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-cycle-count
+</td>
+<td>
+
+The total number of periodic topology-description cleanup cycles run by the
coordinator
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Cleanup Eligible Rate
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-eligible-rate
+</td>
+<td>
+
+The rate of groups found eligible for plugin-state deletion by the cleanup scan
+</td> </tr>
+<tr>
+<td>
+
+Topology Description Cleanup Eligible Count
+</td>
+<td>
+
+kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-eligible-count
+</td>
+<td>
+
+The total number of groups found eligible for plugin-state deletion by the
cleanup scan
+</td> </tr>
+<tr>
+<td>
+
Classic Group Count
</td>
<td>
diff --git a/docs/streams/developer-guide/config-streams.md
b/docs/streams/developer-guide/config-streams.md
index 51bacbc2066..80c13125a9b 100644
--- a/docs/streams/developer-guide/config-streams.md
+++ b/docs/streams/developer-guide/config-streams.md
@@ -1131,6 +1131,23 @@ The maximum amount of time in milliseconds a task might
stall due to internal er
<tr>
<td>
+topology.description.push.enabled
+</td>
+<td>
+
+Medium
+</td>
+<td>
+
+Controls whether the Kafka Streams client sends topology descriptions to the
broker when requested. When set to `false`, the client will not prepare or push
topology descriptions. See [Topology Description
Plugin](/{version}/streams/developer-guide/topology-description-plugin/).
+</td>
+<td>
+
+`true`
+</td> </tr>
+<tr>
+<td>
+
topology.optimization
</td>
<td>
@@ -1691,6 +1708,12 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
> A task assignor class or class name implementing the
> `org.apache.kafka.streams.processor.assignment.TaskAssignor` interface.
> Defaults to the high-availability task assignor. One possible alternative
> implementation provided in Apache Kafka is the
> `org.apache.kafka.streams.processor.assignment.assignors.StickyTaskAssignor`,
> which was the default task assignor before KIP-441 and minimizes task
> movement at the cost of stateful task availability. Alternative
> implementations of the task assi [...]
+ #### topology.description.push.enabled
+
+> Controls whether the Kafka Streams client sends topology descriptions to the
broker when requested. When set to `false`, the client will not prepare or push
topology descriptions. Enabled by default.
+
+This configuration only has an effect for streams groups
(`group.protocol=streams`) on clusters where the broker configuration
`group.streams.topology.description.plugin.class` is set; otherwise, the broker
never requests topology descriptions. See [Topology Description
Plugin](/{version}/streams/developer-guide/topology-description-plugin/) for
details.
+
#### topology.optimization
> A configuration telling Kafka Streams if it should optimize the topology and
> what optimizations to apply. Acceptable values are:
> `StreamsConfig.NO_OPTIMIZATION` (`none`), `StreamsConfig.OPTIMIZE` (`all`)
> or a comma separated list of specific optimizations:
> `StreamsConfig.REUSE_KTABLE_SOURCE_TOPICS` (`reuse.ktable.source.topics`),
> `StreamsConfig.MERGE_REPARTITION_TOPICS` (`merge.repartition.topics`),
> `StreamsConfig.SINGLE_STORE_SELF_JOIN` (`single.store.self.join`).
diff --git a/docs/streams/developer-guide/kafka-streams-group-sh.md
b/docs/streams/developer-guide/kafka-streams-group-sh.md
index cbde1399e50..c8d4b77b197 100644
--- a/docs/streams/developer-guide/kafka-streams-group-sh.md
+++ b/docs/streams/developer-guide/kafka-streams-group-sh.md
@@ -41,7 +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()`.
+ * The processing topology, as recorded by the broker's [topology
description
plugin](/{version}/streams/developer-guide/topology-description-plugin/) (with
`--topology`), in a format that mirrors `Topology#describe()`. Requires brokers
running Apache Kafka 4.4 or newer with
`group.streams.topology.description.plugin.class` configured.
* **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). Internal topics can be deleted by specifying
selected topics with `--delete-internal-topic`, or all internal topics with
`--delete-all-internal-topics`.
@@ -90,6 +90,27 @@ Inspecting group's state, members, and lag
--describe --group my-streams-app --topology
+### Describing the processing topology {#describe-topology}
+
+The `--topology` option prints the processing topology of the group, as
recorded by the broker's [topology description
plugin](/{version}/streams/developer-guide/topology-description-plugin/), in a
format that mirrors `Topology#describe()`:
+
+
+ Topologies:
+ Sub-topology: 0
+ Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])
+ --> KSTREAM-FLATMAPVALUES-0000000001
+ Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
+ --> KSTREAM-AGGREGATE-0000000002
+ <-- KSTREAM-SOURCE-0000000000
+ ...
+
+This requires brokers running Apache Kafka 4.4 or newer with the broker
configuration `group.streams.topology.description.plugin.class` set; against
older brokers the command fails with `UnsupportedVersionException`. If no
topology description is available, the tool prints one of the following
messages and exits with a non-zero exit code:
+
+ * `No topology description is stored for streams group '<id>'.` — No
description is recorded, for example because no topology description plugin is
configured on the broker or the application has not pushed a description yet.
+ * `The broker failed to fetch the topology description for streams group
'<id>'. See the broker logs for details.` — The broker's plugin failed to read
the stored description.
+
+See the [Topology Description
Plugin](/{version}/streams/developer-guide/topology-description-plugin/)
documentation for how the feature works and how to troubleshoot it.
+
## Reset input-topic offsets (preview, then apply) {#reset-offsets}
Ensure all application instances are stopped/inactive. Always preview changes
with `--dry-run` before using `--execute`.
diff --git a/docs/streams/developer-guide/scala-migration.md
b/docs/streams/developer-guide/scala-migration.md
index 0975fce0441..de8a03213a3 100644
--- a/docs/streams/developer-guide/scala-migration.md
+++ b/docs/streams/developer-guide/scala-migration.md
@@ -1,7 +1,7 @@
---
title: Migrating from Streams Scala to Java API
description: Migration guidance from the Kafka Streams Scala API to the Java
API.
-weight: 16
+weight: 17
tags: ['kafka', 'docs']
aliases:
keywords:
diff --git a/docs/streams/developer-guide/streams-rebalance-protocol.md
b/docs/streams/developer-guide/streams-rebalance-protocol.md
index eb49f43dc5d..5f9cc280d58 100644
--- a/docs/streams/developer-guide/streams-rebalance-protocol.md
+++ b/docs/streams/developer-guide/streams-rebalance-protocol.md
@@ -47,6 +47,8 @@ The following features are available in the current release:
* **CLI Integration**: You can list, describe, and delete streams groups via
the
[bin/kafka-streams-groups.sh](/{version}/streams/developer-guide/kafka-streams-group-sh/)
script.
+* **Topology Description Plugin**: Brokers can record a human-readable
description of each streams group's processing topology via a pluggable
backend, configured with `group.streams.topology.description.plugin.class`. The
recorded topology can be inspected via the
[`Admin`](/{version}/javadoc/org/apache/kafka/clients/admin/Admin.html)
interface or `kafka-streams-groups.sh --describe --topology`. See the [Topology
Description Plugin](/{version}/streams/developer-guide/topology-descriptio [...]
+
* **Offline Migration**: After shutting down all members and waiting for their
`session.timeout.ms` to expire (or forcing an explicit group leave), a classic
group can be converted to a streams group and a streams group can be converted
to a classic group. The only broker-side group data that will be preserved are
the committed offsets. Internal topics (changelog and repartition topics) will
continue to exist as regular Kafka topics.
* **Static Membership**: Streams applications can configure
`group.instance.id` when using `group.protocol=streams`. Kafka Streams derives
unique group instance IDs for its stream threads internally.
@@ -117,6 +119,7 @@ The following broker configurations control the behavior of
streams groups. For
*
[`group.streams.num.standby.replicas`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.num.standby.replicas):
The default number of standby replicas for each task.
*
[`group.streams.max.standby.replicas`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.max.standby.replicas):
Maximum for dynamic configurations of the standby replica configuration.
*
[`group.streams.initial.rebalance.delay.ms`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.initial.rebalance.delay.ms):
The first rebalance of a new (ie, previously empty) group is delayed by this
amount to allow more members to join the group.
+*
[`group.streams.topology.description.plugin.class`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.topology.description.plugin.class):
The fully qualified class name of a `StreamsGroupTopologyDescriptionPlugin`
implementation. When not set, the [topology description
feature](/{version}/streams/developer-guide/topology-description-plugin/) is
disabled.
## Group Configuration
@@ -179,6 +182,10 @@ The main differences from consumer group APIs are:
A new tool called `bin/kafka-streams-groups.sh` is added for working with
streams groups. It replaces `bin/kafka-streams-application-reset.sh` for
streams groups and can be used to list, describe, and delete streams groups.
See the [kafka-streams-groups.sh
documentation](/{version}/streams/developer-guide/kafka-streams-group-sh/) for
detailed usage information.
+## Topology Description
+
+When a topology description plugin is configured on the brokers via
`group.streams.topology.description.plugin.class`, the group coordinator
records a human-readable description of each streams group's processing
topology, pushed automatically by the Kafka Streams clients. The description
can be retrieved via `Admin#describeStreamsGroups` or `kafka-streams-groups.sh
--describe --topology`. See the [Topology Description
Plugin](/{version}/streams/developer-guide/topology-description-plugi [...]
+
# Architecture and How It Works
## Streams Groups
diff --git a/docs/streams/developer-guide/topology-description-plugin.md
b/docs/streams/developer-guide/topology-description-plugin.md
new file mode 100644
index 00000000000..a90314770c8
--- /dev/null
+++ b/docs/streams/developer-guide/topology-description-plugin.md
@@ -0,0 +1,195 @@
+---
+title: Topology Description Plugin
+type: docs
+description: Broker-side plugin for recording and exposing the processing
topology of streams groups.
+weight: 16
+tags: ['kafka', 'docs']
+aliases:
+keywords:
+---
+
+<!--
+ 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.
+-->
+
+Starting with Apache Kafka 4.4, brokers can record a human-readable
description of the processing topology of each **streams group**, as defined by
[KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin).
Kafka Streams clients push the same information that `Topology#describe()`
returns to the group coordinator, which hands it to a pluggable, broker-side
storage backend. Operators can then inspect the topology of any streams gro
[...]
+
+# Overview
+
+The feature applies only to streams groups using the [Streams Rebalance
Protocol](/{version}/streams/developer-guide/streams-rebalance-protocol/)
(`group.protocol=streams`, KIP-1071). It is **disabled by default**: the broker
only solicits, stores, and serves topology descriptions when the broker
configuration `group.streams.topology.description.plugin.class` is set to a
`StreamsGroupTopologyDescriptionPlugin` implementation.
+
+When the feature is enabled:
+
+ * Kafka Streams clients automatically push a description of their topology
when the broker requests one — no application code changes are required. The
push can be turned off per client via the Streams configuration
`topology.description.push.enabled`.
+ * The description is versioned by the group's **topology epoch**, so the
broker always knows whether its stored description matches the topology the
group is currently running.
+ * The stored description can be retrieved with `Admin#describeStreamsGroups`
(using `DescribeStreamsGroupsOptions#includeTopologyDescription(true)`) or with
`kafka-streams-groups.sh --describe --topology`.
+
+# How it works: the push/describe cycle
+
+The feature adds one new RPC, `StreamsGroupTopologyDescriptionUpdate`, and
extends the existing `StreamsGroupHeartbeat` and `StreamsGroupDescribe` RPCs.
The cycle works as follows:
+
+ 1. **Solicitation.** When the group coordinator has not yet recorded a
successful push for the group's current topology epoch (for example, a new
group or a topology change that bumped the epoch), it sets the
`TopologyDescriptionRequired` flag in the `StreamsGroupHeartbeat` response.
+ 2. **Push.** A client that sees this flag — and has
`topology.description.push.enabled=true` — sends a
`StreamsGroupTopologyDescriptionUpdate` request to the group coordinator,
containing its group ID, member ID, the topology epoch, and the topology
description (subtopologies, sources, processors, sinks, state stores, and
global stores).
+ 3. **Store.** The broker validates that the sender is a known member of the
group and invokes the plugin's `setTopology(groupId, topologyEpoch,
description)` method. On success, the broker records the stored topology epoch
and stops soliciting. Multiple members may push concurrently for the same
epoch; the pushed data is identical, and the plugin must handle this
idempotently.
+ 4. **Failure handling.** If the plugin fails to store the description, the
broker distinguishes two cases:
+ * A **permanent failure**
(`StreamsTopologyDescriptionPermanentFailureException`) means the description
will never be accepted at this topology epoch (for example, it is too large or
semantically rejected). The broker records the failed epoch and stops
soliciting until the topology epoch advances.
+ * A **transient failure**
(`StreamsTopologyDescriptionTransientFailureException`, or any other exception)
causes the broker to arm a per-group exponential back-off (30 seconds up to 1
hour) and re-solicit the description on a later heartbeat.
+
+ In both cases, the pushing client receives the error code
`STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED`. The client does not retry on its
own; the broker drives retries through heartbeat solicitation.
+ 5. **Describe.** When a caller requests the topology description via
`StreamsGroupDescribe` (version 1 or higher, with
`IncludeTopologyDescription=true`), the broker invokes the plugin's
`getTopology(groupId, topologyEpoch)` method and attaches the description
together with a status field to the response. See [Interpreting the topology
description status](#interpreting-the-topology-description-status) below.
+ 6. **Deletion.** When a streams group is deleted (via `DeleteGroups`) or
expires, the broker invokes the plugin's `deleteTopology(groupId)` method so
that the plugin can clean up its stored data. See [Group deletion and
GROUP_DELETION_FAILED](#group-deletion-and-group_deletion_failed) below for the
failure semantics.
+
+# Broker configuration
+
+ *
[`group.streams.topology.description.plugin.class`](/{version}/configuration/broker-configs#brokerconfigs_group.streams.topology.description.plugin.class):
The fully qualified class name of a `StreamsGroupTopologyDescriptionPlugin`
implementation. When not set (the default), the feature is disabled: the broker
never solicits topology descriptions, and describe requests report status
`NOT_STORED`.
+
+Apache Kafka ships a reference implementation,
`org.apache.kafka.server.streams.InMemoryTopologyDescriptionPlugin`, which
stores one description per group in an in-memory map. It is intended for
testing and as a starting point for real implementations. It is **not suitable
for production** because its state is lost on broker restart and is not shared
across brokers.
+
+# Client configuration
+
+ *
[`topology.description.push.enabled`](/{version}/configuration/kafka-streams-configs#streamsconfigs_topology.description.push.enabled):
Controls whether the Kafka Streams client sends topology descriptions to the
broker when requested. When set to `false`, the client will not prepare or push
topology descriptions. Enabled by default.
+
+Note that this configuration only controls whether the client *responds* to
broker solicitations. If the broker has no plugin configured, the client is
never asked to push, regardless of this setting.
+
+# Implementing a plugin
+
+A plugin implements the
[`StreamsGroupTopologyDescriptionPlugin`](/{version}/javadoc/org/apache/kafka/coordinator/group/api/streams/StreamsGroupTopologyDescriptionPlugin.html)
interface from the `group-coordinator-api` module:
+
+```java
+public interface StreamsGroupTopologyDescriptionPlugin extends Configurable,
AutoCloseable {
+
+ CompletableFuture<Void> setTopology(String groupId, int topologyEpoch,
StreamsGroupTopologyDescription description);
+
+ CompletableFuture<Void> deleteTopology(String groupId);
+
+ CompletableFuture<StreamsGroupTopologyDescription> getTopology(String
groupId, int topologyEpoch);
+}
+```
+
+Guidelines for implementations:
+
+ * **Be thread-safe.** `setTopology` may be called concurrently by multiple
members of the same group.
+ * **Be idempotent.** Calls to `setTopology` with the same `(groupId,
topologyEpoch)` carry identical data and must be idempotent. `deleteTopology`
may be called more than once for the same group, including when nothing is
stored.
+ * **Complete futures asynchronously; never throw synchronously.** Failures
must be signalled by completing the returned future exceptionally. A
synchronous throw from `setTopology` is treated as a permanent failure with a
generic client-visible error message.
+ * **Classify failures.** Complete the `setTopology` future with
`StreamsTopologyDescriptionPermanentFailureException` when the description will
never be accepted at this topology epoch, and with
`StreamsTopologyDescriptionTransientFailureException` (or any other exception)
for retriable backend failures. The permanent-vs-transient distinction is
broker-internal; the pushing client always sees
`STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` with the exception's message.
+ * **Key data by group and epoch.** `getTopology(groupId, topologyEpoch)`
should return the description only if it matches the requested topology epoch,
and complete with `null` when the plugin no longer has the data (for example,
after a backend wipe) — the broker then reports status `NOT_STORED`. If the
future completes exceptionally, the broker reports a read error (status
`ERROR`) for the group. Note that the broker only solicits a new push when it
has not recorded a successful push [...]
+ * **Lifecycle.** The plugin is instantiated once per broker, configured via
`Configurable#configure(Map)` with the broker configuration, and closed via
`AutoCloseable#close()` on broker shutdown.
+
+# Reading the topology description
+
+## Admin API
+
+Pass `DescribeStreamsGroupsOptions#includeTopologyDescription(true)` to
`Admin#describeStreamsGroups`:
+
+```java
+try (Admin admin = Admin.create(props)) {
+ DescribeStreamsGroupsResult result = admin.describeStreamsGroups(
+ List.of("my-streams-app"),
+ new DescribeStreamsGroupsOptions().includeTopologyDescription(true));
+ StreamsGroupDescription description =
result.describedGroups().get("my-streams-app").get();
+ StreamsGroupTopologyDescriptionStatus status =
description.topologyDescriptionStatus();
+ Optional<StreamsGroupTopologyDescription> topology =
description.topologyDescription();
+}
+```
+
+The returned `StreamsGroupTopologyDescription` mirrors
`org.apache.kafka.streams.TopologyDescription` (subtopologies with source,
processor, and sink nodes, plus global stores) without requiring a dependency
on the `kafka-streams` library. Requesting a topology description against a
broker that does not support it (older than 4.4) fails with
`UnsupportedVersionException`.
+
+## CLI
+
+Use the `--topology` option of `bin/kafka-streams-groups.sh` together with
`--describe`:
+
+```
+kafka-streams-groups.sh --bootstrap-server localhost:9092 \
+ --describe --group my-streams-app --topology
+```
+
+When a description is available, the output mirrors the format of
`Topology#describe()`:
+
+```
+Topologies:
+ Sub-topology: 0
+ Source: KSTREAM-SOURCE-0000000000 (topics: [streams-plaintext-input])
+ --> KSTREAM-FLATMAPVALUES-0000000001
+ Processor: KSTREAM-FLATMAPVALUES-0000000001 (stores: [])
+ --> KSTREAM-AGGREGATE-0000000002
+ <-- KSTREAM-SOURCE-0000000000
+ Processor: KSTREAM-AGGREGATE-0000000002 (stores: [counts-store])
+ --> KSTREAM-SINK-0000000003
+ <-- KSTREAM-FLATMAPVALUES-0000000001
+ Sink: KSTREAM-SINK-0000000003 (topic: streams-wordcount-output)
+ <-- KSTREAM-AGGREGATE-0000000002
+```
+
+If no description is available, the tool prints an explanatory message and
exits with a non-zero exit code. See the [kafka-streams-groups.sh
documentation](/{version}/streams/developer-guide/kafka-streams-group-sh/) for
the full CLI reference.
+
+## Interpreting the topology description status
+
+Every describe response that requested a topology description carries a
`StreamsGroupTopologyDescriptionStatus`. The description itself is present if
and only if the status is `AVAILABLE`.
+
+ * `NOT_REQUESTED`: The topology description was not requested (the caller
did not set `includeTopologyDescription(true)`).
+ * `NOT_STORED`: No topology description is recorded for this group — for
example, because no topology description plugin is configured on the broker, or
the clients have not pushed a description yet.
+ * `ERROR`: The broker failed to fetch the topology description from the
plugin. See the broker logs for details.
+ * `AVAILABLE`: The topology description is available and carried in the
response.
+
+# Group deletion and GROUP_DELETION_FAILED
+
+When a streams group is deleted while a topology description plugin is
configured, the broker calls the plugin's `deleteTopology` method before
removing the group. If the plugin fails to delete its data, the `DeleteGroups`
request returns the error code `GROUP_DELETION_FAILED` for that group, with the
plugin's exception message in the per-group `ErrorMessage` field (available in
`DeleteGroups` version 3 and higher), and the broker does **not** delete the
group. Retrying the deletion re-i [...]
+
+# Observability
+
+The broker exposes metrics for every plugin interaction under the MBean group
`kafka.server:type=group-coordinator-metrics`; the full list is in the [group
coordinator monitoring
reference](/{version}/operations/monitoring#group-coordinator-monitoring). Each
sensor is published as both a `-rate` (per-second) and a `-count` (cumulative)
metric, so `streams-group-topology-description-set-success` becomes
`streams-group-topology-description-set-success-rate` and
`streams-group-topology-desc [...]
+
+ * `streams-group-topology-description-set-success` /
`streams-group-topology-description-set-error`: outcomes of `setTopology`
calls, driven by client pushes.
+ * `streams-group-topology-description-get-success` /
`streams-group-topology-description-get-error`: outcomes of `getTopology`
calls, driven by describe requests.
+ * `streams-group-topology-description-delete-success` /
`streams-group-topology-description-delete-error`: outcomes of `deleteTopology`
calls, driven by group deletion and cleanup.
+ * `streams-group-topology-description-cleanup-cycle`: number of periodic
cleanup cycles the coordinator has run.
+ * `streams-group-topology-description-cleanup-eligible`: number of groups
the cleanup scan found eligible for plugin-state deletion.
+
+Watch the `-error` sensors first: a rising `get-error` rate explains `ERROR`
describe responses, a rising `set-error` rate explains descriptions that never
appear, and a rising `delete-error` rate explains `GROUP_DELETION_FAILED`.
+
+# Troubleshooting
+
+Before reading broker logs, check the
`streams-group-topology-description-*-error` metrics described under
[Observability](#observability) — they pinpoint which plugin call (`set`,
`get`, or `delete`) is failing.
+
+**`--topology` reports "No topology description is stored" (status
`NOT_STORED`).**
+
+ * Verify that `group.streams.topology.description.plugin.class` is set on
all brokers hosting the group coordinator. Without it, the feature is disabled.
+ * Verify that the application does not set
`topology.description.push.enabled=false`.
+ * If the group (or its topology epoch) is new, the clients may simply not
have pushed yet — the broker solicits the push via the heartbeat, so the
description typically appears within a few heartbeat intervals.
+ * If the description still does not appear, check the broker logs for failed
`setTopology` calls. After a permanent failure (for example, a description the
plugin rejects), the broker stops soliciting until the topology epoch advances.
+ * If the description used to appear and has now vanished, the plugin may
have lost its stored data. When the plugin's `getTopology` returns `null`, the
broker surfaces the status as `NOT_STORED` (logged at `WARN`) and keeps
returning `NOT_STORED` on subsequent describes. Because the broker only
re-solicits a push when the topology epoch advances, restarting the application
without bumping the topology will not recover the description — advance the
topology epoch or clear the plugin sta [...]
+
+**Status `ERROR` when describing.**
+
+ * The plugin's `getTopology` call failed on the broker. Check the broker
logs of the group coordinator for the underlying exception.
+
+**Push delivery issues.**
+
+ * A failed push surfaces to the client as
`STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` and is logged by the Streams
client; the client does not retry on its own. The broker re-solicits the push
via the heartbeat — after transient plugin failures with an exponential
back-off between 30 seconds and 1 hour.
+ * If the pushing member has been fenced or the group was deleted meanwhile,
the push fails with `UNKNOWN_MEMBER_ID` and the client rejoins the group; this
is expected and self-healing.
+
+**`DeleteGroups` fails with `GROUP_DELETION_FAILED`.**
+
+ * The plugin failed to delete the stored description. The per-group error
message contains the plugin's failure reason; the broker logs contain the full
exception. Resolve the plugin/backend problem and retry the deletion —
`deleteTopology` is invoked again idempotently.
+
+**`UnsupportedVersionException` when requesting the topology description.**
+
+ * The broker is older than Apache Kafka 4.4 and does not support
`StreamsGroupDescribe` version 1. Upgrade the broker, or describe the group
without requesting the topology description.
+
+ * [Documentation](/documentation)
+ * [Kafka Streams](/documentation/streams)
+ * [Developer Guide](/documentation/streams/developer-guide/)
+
diff --git a/docs/streams/upgrade-guide.md b/docs/streams/upgrade-guide.md
index df78b93d72c..c6e737f56c7 100644
--- a/docs/streams/upgrade-guide.md
+++ b/docs/streams/upgrade-guide.md
@@ -71,6 +71,8 @@ Kafka Streams no longer emits a WARN from
`KafkaStreams#cleanUp()` when the appl
Kafka Streams now validates the `application.server` configuration when
`StreamsConfig` is created. The value must be empty or a valid endpoint from
which Kafka Streams can parse both host and port, such as `host:port` or
`protocol://host:port`. Invalid values that may previously have failed later
during startup or assignment now fail earlier with a `ConfigException`. More
details can be found in
[KIP-1245](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1245%3A+Enforce+%27applicat
[...]
+For applications using the Streams Rebalance Protocol
(`group.protocol=streams`), brokers can now record a human-readable description
of the group's processing topology via a pluggable backend
([KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin)).
When the broker configuration
`group.streams.topology.description.plugin.class` is set, Kafka Streams clients
automatically push a description equivalent to `Topology#describe()` t [...]
+
## Streams API changes in 4.3.0
**Note:** Kafka Streams 4.3.0 contains a critical native memory leak in the
RocksDB state store layer
([KAFKA-20616](https://issues.apache.org/jira/browse/KAFKA-20616)). The
`ColumnFamilyOptions` for the offsets column family is not closed, and column
family handles can leak on close-path exceptions, which under cascading task
closes (e.g., rebalances or error-triggered recoveries) leads to unbounded
off-heap memory growth and eventual OOM. Users running Kafka Streams should
consider upg [...]
