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. 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 group via the Admin API or the bin/kafka-streams-groups.sh CLI — without access to the application's source code or a running instance.
The feature applies only to streams groups using the 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:
topology.description.push.enabled.Admin#describeStreamsGroups (using DescribeStreamsGroupsOptions#includeTopologyDescription(true)) or with kafka-streams-groups.sh --describe --topology.The feature adds one new RPC, StreamsGroupTopologyDescriptionUpdate, and extends the existing StreamsGroupHeartbeat and StreamsGroupDescribe RPCs. The cycle works as follows:
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.
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).
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.
Failure handling. If the plugin fails to store the description, the broker distinguishes two cases:
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.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.
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 below.
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 below for the failure semantics.
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.
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.
A plugin implements the StreamsGroupTopologyDescriptionPlugin interface from the group-coordinator-api module:
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:
setTopology may be called concurrently by multiple members of the same group.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.setTopology is treated as a permanent failure with a generic client-visible error message.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.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 for the current topology epoch; a plugin that loses already-stored data is not automatically re-populated until the topology epoch advances, so implementations should use durable storage.Configurable#configure(Map) with the broker configuration, and closed via AutoCloseable#close() on broker shutdown.Pass DescribeStreamsGroupsOptions#includeTopologyDescription(true) to Admin#describeStreamsGroups:
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.
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 for the full CLI reference.
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.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-invokes deleteTopology idempotently. Groups that expire through periodic cleanup are treated identically — their removal is deferred to a future cleanup cycle until the plugin deletion succeeds.
This cleanup guarantee holds as long as the same plugin stays configured on the broker that owns the group. It relies on the broker‘s own group-deletion and periodic-cleanup logic to invoke deleteTopology; if that logic stops running against a given plugin instance, entries in the plugin’s storage are left behind.
Turning off the plugin (unsetting group.streams.topology.description.plugin.class) or downgrading the broker to a version that predates KIP-1331 both stop the broker from calling deleteTopology for groups that already have data recorded. Those entries are left behind in the plugin's backing storage — the broker has no way to reach them once it stops running the plugin.
This is expected, not a sign of corruption: the plugin‘s storage is external to the broker, so removing or downgrading the component that manages its lifecycle naturally leaves it self-managed from then on. If you plan to remove the plugin permanently or downgrade for good, clean up the plugin’s backing storage yourself (however your plugin implementation exposes that — a table truncate, a key prefix delete, and so on) rather than relying on the broker to do it after the fact. A brief downgrade that gets rolled forward again does not need any cleanup: once the plugin is reconfigured and the broker upgraded back, normal pushes and cleanup cycles resume and reconcile the group's state with the plugin as usual.
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. 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-description-set-success-count.
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.
Before reading broker logs, check the streams-group-topology-description-*-error metrics described under Observability — they pinpoint which plugin call (set, get, or delete) is failing.
--topology reports “No topology description is stored” (status NOT_STORED).
group.streams.topology.description.plugin.class is set on all brokers hosting the group coordinator. Without it, the feature is disabled.topology.description.push.enabled=false.setTopology calls. After a permanent failure (for example, a description the plugin rejects), the broker stops soliciting until the topology epoch advances.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 state to trigger a fresh push.Status ERROR when describing.
getTopology call failed on the broker. Check the broker logs of the group coordinator for the underlying exception.Push delivery issues.
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.UNKNOWN_MEMBER_ID and the client rejoins the group; this is expected and self-healing.DeleteGroups fails with GROUP_DELETION_FAILED.
deleteTopology is invoked again idempotently.UnsupportedVersionException when requesting the topology description.
StreamsGroupDescribe version 1. Upgrade the broker, or describe the group without requesting the topology description.Leftover entries in the plugin's storage after removing the plugin or downgrading.
See Removing the plugin or downgrading the broker — this is expected, and the storage needs to be cleaned up manually.