From b53de049d7eb1ac65eab8637926ede1c49b37bbe Mon Sep 17 00:00:00 2001 From: aliehsaeedii Date: Tue, 7 Jul 2026 12:42:04 +0200 Subject: [PATCH 1/2] KAFKA-20630: Add documentation for topology description plugin feature Documents the streams group topology description plugin (KIP-1331): - New developer-guide page (topology-description-plugin.md) covering the feature overview, the push/describe cycle, broker and client configuration, plugin implementation guidelines, the Admin API and CLI surface, TopologyDescriptionStatus interpretation, GROUP_DELETION_FAILED semantics, and troubleshooting. - config-streams.md: add topology.description.push.enabled to the config table and detailed entries. - kafka-streams-group-sh.md: document --topology output format, broker requirements, and error messages. - streams-rebalance-protocol.md: list the feature as supported, add the new broker config, and link the new page from the administration section. - getting-started/upgrade.md: notable-changes entry for 4.4.0 covering the new RPC, RPC version bumps (StreamsGroupDescribe/Heartbeat v1, DeleteGroups v3), new error codes, and new 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. Co-Authored-By: Claude Fable 5 --- docs/getting-started/upgrade.md | 1 + docs/operations/basic-kafka-operations.md | 2 +- .../streams/developer-guide/config-streams.md | 23 +++ .../developer-guide/kafka-streams-group-sh.md | 23 ++- .../developer-guide/scala-migration.md | 2 +- .../streams-rebalance-protocol.md | 7 + .../topology-description-plugin.md | 180 ++++++++++++++++++ docs/streams/upgrade-guide.md | 2 + 8 files changed, 237 insertions(+), 3 deletions(-) create mode 100644 docs/streams/developer-guide/topology-description-plugin.md diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md index f2594cdd3e89e..bd429ee95b617 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.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 diff --git a/docs/operations/basic-kafka-operations.md b/docs/operations/basic-kafka-operations.md index 4ed37d87417ee..d5146f7404d91 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-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 diff --git a/docs/streams/developer-guide/config-streams.md b/docs/streams/developer-guide/config-streams.md index 51bacbc20663a..80c13125a9b36 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 +topology.description.push.enabled + + + +Medium + + + +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/). + + + +`true` + + + + topology.optimization @@ -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`). diff --git a/docs/streams/developer-guide/kafka-streams-group-sh.md b/docs/streams/developer-guide/kafka-streams-group-sh.md index cbde1399e50aa..c8d4b77b19744 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 ''.` — 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 ''. 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 0975fce04416e..de8a03213a3a3 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 eb49f43dc5ded..5f9cc280d58eb 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-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. @@ -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-plugin/) documentation for the push/describe workflow, plugin implementation guidelines, and troubleshooting. + # 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 0000000000000..2702384d896cf --- /dev/null +++ b/docs/streams/developer-guide/topology-description-plugin.md @@ -0,0 +1,180 @@ +--- +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: +--- + + + +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 group via the [`Admin`](/{version}/javadoc/org/apache/kafka/clients/admin/Admin.html) API or the `bin/kafka-streams-groups.sh` CLI — without access to the application's source code or a running instance. + +# 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 setTopology(String groupId, int topologyEpoch, StreamsGroupTopologyDescription description); + + CompletableFuture deleteTopology(String groupId); + + CompletableFuture 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 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. + * **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 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-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. + +# Troubleshooting + +**`--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. + +**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 df78b93d72c5f..c6e737f56c7a4 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+%27application.server%27+%3Cserver%3E%3A%3Cport%3E+format+at+config+level). +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()` to the group coordinator, and operators can retrieve it via `Admin#describeStreamsGroups` (with `DescribeStreamsGroupsOptions#includeTopologyDescription(true)`) or `kafka-streams-groups.sh --describe --topology` — without access to the application's source code. The push can be disabled per client via the new configuration `topology.description.push.enabled` (default `true`). More details can be found in the [Topology Description Plugin](/{version}/streams/developer-guide/topology-description-plugin/) documentation. + ## 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 upgrading directly to 4.3.1, which includes the fix for it. From ccc0b2353f44efcdbc20b2acd69902acab7ef93f Mon Sep 17 00:00:00 2001 From: aliehsaeedii Date: Mon, 13 Jul 2026 17:43:39 +0200 Subject: [PATCH 2/2] address comments --- docs/operations/monitoring.md | 208 ++++++++++++++++++ .../topology-description-plugin.md | 15 ++ 2 files changed, 223 insertions(+) diff --git a/docs/operations/monitoring.md b/docs/operations/monitoring.md index 3f5fb0ec66f40..92badfae27aaf 100644 --- a/docs/operations/monitoring.md +++ b/docs/operations/monitoring.md @@ -1348,6 +1348,214 @@ Total number of Streams Group Rebalances +Topology Description Set Success Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-success-rate + + + +The rate of successful setTopology plugin calls (driven by client topology-description pushes) + + + + +Topology Description Set Success Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-success-count + + + +The total number of successful setTopology plugin calls + + + + +Topology Description Set Error Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-error-rate + + + +The rate of failed setTopology plugin calls + + + + +Topology Description Set Error Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-set-error-count + + + +The total number of failed setTopology plugin calls + + + + +Topology Description Get Success Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-success-rate + + + +The rate of successful getTopology plugin calls (driven by describe requests) + + + + +Topology Description Get Success Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-success-count + + + +The total number of successful getTopology plugin calls + + + + +Topology Description Get Error Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-error-rate + + + +The rate of failed getTopology plugin calls + + + + +Topology Description Get Error Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-get-error-count + + + +The total number of failed getTopology plugin calls + + + + +Topology Description Delete Success Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-success-rate + + + +The rate of successful deleteTopology plugin calls (driven by group deletion and cleanup) + + + + +Topology Description Delete Success Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-success-count + + + +The total number of successful deleteTopology plugin calls + + + + +Topology Description Delete Error Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-error-rate + + + +The rate of failed deleteTopology plugin calls + + + + +Topology Description Delete Error Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-delete-error-count + + + +The total number of failed deleteTopology plugin calls + + + + +Topology Description Cleanup Cycle Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-cycle-rate + + + +The rate of periodic topology-description cleanup cycles run by the coordinator + + + + +Topology Description Cleanup Cycle Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-cycle-count + + + +The total number of periodic topology-description cleanup cycles run by the coordinator + + + + +Topology Description Cleanup Eligible Rate + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-eligible-rate + + + +The rate of groups found eligible for plugin-state deletion by the cleanup scan + + + + +Topology Description Cleanup Eligible Count + + + +kafka.server:type=group-coordinator-metrics,name=streams-group-topology-description-cleanup-eligible-count + + + +The total number of groups found eligible for plugin-state deletion by the cleanup scan + + + + Classic Group Count diff --git a/docs/streams/developer-guide/topology-description-plugin.md b/docs/streams/developer-guide/topology-description-plugin.md index 2702384d896cf..a90314770c8f2 100644 --- a/docs/streams/developer-guide/topology-description-plugin.md +++ b/docs/streams/developer-guide/topology-description-plugin.md @@ -148,14 +148,29 @@ Every describe response that requested a topology description carries a `Streams 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. +# 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-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`. + # 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 state to trigger a fresh push. **Status `ERROR` when describing.**