Skip to content

KAFKA-20630: Add documentation for topology description plugin feature #22776

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
frankvicky merged 2 commits into from
Jul 14, 2026
Merged

KAFKA-20630: Add documentation for topology description plugin feature #22776

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b53de04
KAFKA-20630: Add documentation for topology description plugin feature
aliehsaeedii Jul 7, 2026
ccc0b23
address comments
aliehsaeedii Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/getting-started/upgrade.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.allow.unverified.issuer` configs (both default `false`) to `true`.
* 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.enabled` (default `true`) controls whether the client pushes topology descriptions when requested. This adds a new RPC, `StreamsGroupTopologyDescriptionUpdate`, bumps `StreamsGroupDescribe` and `StreamsGroupHeartbeat` to version 1, and introduces the error codes `GROUP_DELETION_FAILED` (134) and `STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED` (135). `DeleteGroups` is bumped to version 3, adding a per-group `ErrorMessage` field so brokers can report why a group deletion failed (for example, when the plugin fails to delete its stored topology description, the group is not deleted and `GROUP_DELETION_FAILED` is returned; retrying the deletion is safe). For further details, please refer to [KIP-1331](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1331%3A+Streams+Group+Topology+Description+Plugin) and the [Topology Description Plugin](/{version}/streams/developer-guide/topology-description-plugin/) documentation.

## Upgrading to 4.3.0

Expand Down
2 changes: 1 addition & 1 deletion docs/operations/basic-kafka-operations.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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-description-plugin/) is configured on the brokers), and delete streams groups. For example, to list all groups in the cluster:

```bash
$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list
Expand Down
208 changes: 208 additions & 0 deletions docs/operations/monitoring.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
23 changes: 23 additions & 0 deletions docs/streams/developer-guide/config-streams.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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 assignment algorithm can be plugged into the application by implementing a custom `TaskAssignor` and setting this config to the name of the custom task assignor class.

#### 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`).
Expand Down
23 changes: 22 additions & 1 deletion docs/streams/developer-guide/kafka-streams-group-sh.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
Expand Down Expand Up @@ -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`.
Expand Down
2 changes: 1 addition & 1 deletion docs/streams/developer-guide/scala-migration.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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:
Expand Down
7 changes: 7 additions & 0 deletions docs/streams/developer-guide/streams-rebalance-protocol.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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-description-plugin/) documentation for details.

* **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.
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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-plugin/) documentation for the push/describe workflow, plugin implementation guidelines, and troubleshooting.

# Architecture and How It Works

## Streams Groups
Expand Down
Loading
Loading