...

package org.apache.kafka.coordinator.group.api.streams;

public class StreamsGroupTopologyDescription {
    public Collection<Subtopology> subtopologies();
    public Collection<GlobalStore> globalStores();

    public static class Subtopology {
        public String id();
        public Collection<Node> nodes();
    }

	/**
     * A processing node in the topology. Predecessor nodes can be inferred from successor relation.
     */
    public interface Node {
        String name();
        Set<String> successors();
    }

    public static class Source implements Node {
        public Set<String> topics();
    }

    public static class Processor implements Node {
        public Set<String> stores();
    }

    public static class Sink implements Node {
        public Optional<String> topic();
    }

    public static class GlobalStore {
        public Source source();
        public Processor processor();
    }
}

Broker-Side Persistence

Two new tagged fields are added to the persisted StreamsGroupMetadataValue record (a broker-internal record written to __consumer_offsets, not a wire-level change visible to clients):

...

Exit code. The command exits 0 for AVAILABLE. It exits 1 for NOT_STORED, for ERROR, and when the DescribedGroup.ErrorCode is non-zero (authorization failure, group not found, coordinator unavailable, etc.). NOT_REQUESTED does not occur for this command because it always sets IncludeTopologyDescription=true.

Proposed Changes

End-to-end Flow

The sequence below traces the four user-visible interactions — a successful push, a describe, an explicit DeleteGroups, and broker-driven cleanup of a naturally-expired group. Plugin calls return CompletableFutures; the broker awaits them asynchronously.

Image Added

Broker Side

1. After a successful StreamsGroupHeartbeat, the broker decides whether to set TopologyDescriptionRequired=true purely from the group's persisted state — no plugin RPC is involved. STALE_TOPOLOGY members are skipped. For all other members, the broker sets the flag iff: StoredTopologyEpoch != currentTopologyEpoch AND LastFailedTopologyEpoch != currentTopologyEpoch AND no per-group transient-failure back-off is currently in its window. The back-off is in-memory state on the service (keyed by groupId, carrying topologyEpoch + nextAttemptMs), armed when a transient setTopology failure is observed and doubled per consecutive failure starting at 30 s and capped at 1 h; it is cleared on a successful push, on a permanent failure (because LastFailedTopologyEpoch ratchets the same epoch), and implicitly on any topology-epoch advance (a stale entry for an older epoch is ignored). A service-side in-flight tracker (per groupId, default 30 s) is additionally used to prevent multiple concurrent heartbeats from each setting the flag.
2. On UpdateStreamsGroupTopologyDescription, the broker checks the READ ACL on the group and that a plugin is configured. The broker validates the MemberId against the streams group: an empty MemberId is rejected with INVALID_REQUEST, a MemberId that does not match any current member of the group is rejected with UNKNOWN_MEMBER_ID, and a request whose group ID does not name an existing streams group is also rejected with UNKNOWN_MEMBER_ID (the group-deleted case is observationally identical to a member fence from the client's point of view, and is handled by the same rejoin path). The broker does not enforce a size limit on the topology description — it is the plugin's responsibility to decide what size it is willing to store. The broker calls setTopology on the plugin. On success the broker writes a metadata record setting StoredTopologyEpoch = pushedEpoch, and the response carries NONE. On InvalidRequestException (mapped to INVALID_REQUEST) or TopologyDescriptionTooLargeException (mapped to TOPOLOGY_DESCRIPTION_TOO_LARGE), the broker writes a metadata record setting LastFailedTopologyEpoch = pushedEpoch so subsequent heartbeats at the same topology epoch do not re-solicit. Any other exception maps to TOPOLOGY_DESCRIPTION_UPDATE_FAILED, is logged at WARN, and is treated as transient — no metadata record is written, and the next heartbeat re-solicits. If the plugin call succeeds but the subsequent metadata-record write fails, the broker accepts the drift: the next heartbeat sees StoredTopologyEpoch < currentTopologyEpoch, re-solicits, the client re-pushes the identical payload, the plugin's idempotent setTopology is invoked again, and the metadata-record write is retried — closing the gap.
3. On DeleteGroups, the broker calls deleteTopology on the plugin before writing the group tombstone, for each requested streams group that has StoredTopologyEpoch != -1. deleteTopology failures are logged but do not affect the deletion response. The group is then tombstoned regardless of the per-group plugin outcome. This ordering matches the broker-driven natural-expiration cleanup (next bullet), where the plugin is also called before the group is tombstoned.
4. On StreamsGroupDescribe with IncludeTopologyDescription=true, the broker calls getTopology on the plugin only when StoredTopologyEpoch == currentTopologyEpoch for that group; otherwise it reports NOT_STORED without making a plugin call. Calls across groups run in parallel. Authorization is unchanged — the existing DESCRIBE ACL on the GROUP resource covers the topology description. DescribedGroup.ErrorCode is never modified by topology-related outcomes; the TopologyDescriptionStatus field carries the reason when TopologyDescription is null (NOT_REQUESTED, NOT_STORED, or ERROR; the last is also logged at WARN). When no plugin is configured, the broker returns NOT_STORED. If getTopology returns null while the broker believed the description was stored (plugin data-loss), the broker schedules a fire-and-forget write resetting StoredTopologyEpoch = -1 so the next heartbeat re-solicits a fresh push.
5. The When a plugin is configured, the broker runs a periodic topology-description cleanup every every offsets.retention.check.interval.ms. Each fire fans out a read-only query across the broker's hosted __consumer_offsets partitions to identify streams groups eligible for cleanup: isEmpty && allOffsetsExpired && StoredTopologyEpoch != -1. This is the same eligibility predicate the shard's offset-expiration sweep uses to delete consumer/share groups, with the additional StoredTopologyEpoch != -1 filter. For each eligible group, the broker calls plugin.deleteTopology(groupId) and, on success, writes a metadata record setting StoredTopologyEpoch = -1. Plugin failures leave the field set; the same group is retried on the next cycle. Once StoredTopologyEpoch = -1, the shard's offset-expiration sweep tombstones the (now flag-cleared) group on a subsequent cycle.

Client Side

1. When no plugin is configured, the periodic cleanup does not run and the shard's offset-expiration sweep expires streams groups normally, ignoring StoredTopologyEpoch; operators that disable a previously-configured plugin are responsible for cleaning up plugin-side state out-of-band.

Client Side

1. At startup, if topology.description.push.enabled=true, the Streams client converts the

2. The Streams client records the TopologyDescriptionRequired flag from each heartbeat response.

3. At startup, if topology.description.push.enabled=true, the Streams client converts the topology returned by Topology#describe() to the wire format and stores it internally. The mapping is one-to-one with the RPC schema; predecessor edges are not sent on the wire and the read side reconstructs them by inverting each node's successor list. When topology.description.push.enabled=false, no description is stored and the feature is disabled on this client.
4. The Streams client records the TopologyDescriptionRequired flag from each heartbeat response.
5. On each consumer background-thread poll, the client sends sends UpdateStreamsGroupTopologyDescription to the coordinator when a coordinator is known, the flag is set, a stored topology description is available, the client has a non-empty member ID assigned by the coordinator, and no prior request is in flight. The member ID populated on the request is the same one carried on StreamsGroupHeartbeat. Completion handling is described in Error Handling and Retries. The push runs on the consumer background thread and never blocks user-facing Kafka Streams APIs

...

Topology Translation

TopologyDescription from Kafka Streams maps one-to-one to the wire types in the RPC schema. Predecessor edges are not sent on the wire; the read side reconstructs them by inverting each node's successor list.

Error Handling and Retries

1. ; the push is best-effort

...

1. .
2. Completion handling on the push response is keyed on the error code.

1. NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery

...

1. and leave the flag

...

1. set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set for retry on the next poll.

...

1. UNKNOWN_MEMBER_ID means the broker no longer recognizes this member (group deleted, or member dropped): the client clears the flag and relies on the existing membership-management path to trigger a clean rejoin on the next heartbeat. All other errors (TOPOLOGY_DESCRIPTION_TOO_LARGE, TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the

...

1. flag and log at WARN

...

1. ; the client does not retry on its own

...

1. , and a re-

...

1. attempt happens only when the broker re-sets the flag via a subsequent heartbeat.

...

1. A non-zero ThrottleTimeMs on the response delays the next push attempt by that amount, as with other request managers.

Plugin Implementation Guidelines

...

A correct plugin implementation should:

• Treat setTopology (on (groupId, topologyEpoch)) and deleteTopology (on (groupId)) as idempotent; the broker may re-issue an identical call when an earlier call's bookkeeping write failed.
• Be thread-safe under concurrent invocation: setTopology may be called by multiple members of the same group in the same heartbeat cycle, and the periodic-cleanup path may invoke deleteTopology while a member is mid-push.
• Reject payloads the plugin will not accept by completing the setTopology future with TopologyDescriptionTooLargeException or InvalidRequestException; the broker persists the rejection at the epoch level via LastFailedTopologyEpoch and stops re-soliciting at the same epoch.
• Signal transient storage-layer failures by completing the setTopology future with any other exception; the broker's transient-failure back-off (30 s → 1 h, exponential) throttles re-solicitation.
• Return null from getTopology when the plugin has lost its data; the broker clears StoredTopologyEpoch on the next describe so a fresh push is re-solicited.
• Avoid blocking coordinator threads (plugin methods may be invoked on them) and complete futures within seconds, not minutes; the broker applies no wall-clock deadline to plugin calls, so the coordinator's responsiveness is bounded by what the plugin does. Bound plugin-side state explicitly — the plugin shares the broker heap

UNKNOWN_MEMBER_ID means the broker no longer recognizes this member — either the group has been deleted or the member has been dropped. The client clears its topologyDescriptionRequired flag and relies on the existing membership-management path: the next StreamsGroupHeartbeat returns the same fence error, which already triggers a clean rejoin. Clearing the flag here prevents another push at the (now-fenced) member ID before the heartbeat round-trips.

A non-zero ThrottleTimeMs on the response delays the next push attempt by that amount, as with other request managers.

Plugin Implementation Guidelines

A correct plugin implementation should:

• Treat setTopology as idempotent on (groupId, topologyEpoch): the broker may re-issue an identical call when an earlier call's bookkeeping write failed. Overwriting with identical data is safe and expected.
• Treat deleteTopology as idempotent on (groupId): the broker may call it again when an earlier call's flag-clearing write failed, and may call it for a group with nothing currently stored. Both must succeed.
• Decide and enforce a maximum stored description size, and reject pushes that exceed it by completing the setTopology future with TopologyDescriptionTooLargeException. The broker persists the rejection at the epoch level via LastFailedTopologyEpoch, so subsequent heartbeats at the same epoch do not re-solicit. The plugin does not need to maintain its own "disabled" tracking.
• Reject payloads with semantic problems (malformed graph, missing fields the plugin requires for its own indexing, etc.) by completing the setTopology future with InvalidRequestException. The broker's permanent-failure treatment matches TopologyDescriptionTooLargeException.
• Surface storage-layer failures by completing the setTopology future with any other exception. The broker treats it as transient: the metadata-record write is skipped, and the next heartbeat re-solicits. No special back-off is required on the plugin side.
• Detect plugin-side data loss by returning null from getTopology when the broker asks for an epoch the plugin no longer has. The broker self-heals by clearing StoredTopologyEpoch on the next describe.

Operational expectations

Plugin code runs inside the broker JVM and shares its heap and threads. The broker applies no wall-clock deadline to plugin calls — matching the convention established by Authorizer and ClientMetricsReceiver — so the operational behaviour of the broker is bounded by what the plugin does, not by a defensive timer. The expectations below are the contract that makes that arrangement safe; plugins that violate them can degrade or stall the coordinator.

• Latency. Plugin futures should settle in seconds, not minutes. A setTopology, getTopology, or deleteTopology future that hangs holds the corresponding broker bookkeeping (response future, in-flight tracker entry, retained payload) for as long as it remains incomplete. If the backing storage is slow or unresponsive, complete the future exceptionally rather than holding it open — the broker's transient-failure back-off (exponential, 30 s → 1 h) will throttle re-solicitation appropriately.
• No blocking on coordinator threads. Plugin methods may be invoked on coordinator threads. Synchronous I/O against the plugin's backing store, locks held across await(), or long-running computation inside setTopology / getTopology / deleteTopology will block coordinator processing for other groups on the same shard. Use the async I/O surface of the chosen backend.
• Bounded memory. The plugin shares the broker heap. Plugin-side state should be bounded to roughly one topology per active group at the current topology epoch; anything larger (caches of superseded epochs, retained payloads after deleteTopology) needs an explicit eviction policy.
• Thread hygiene. setTopology may be called concurrently by multiple members of the same group in the same heartbeat cycle, and the periodic-cleanup path may invoke deleteTopology while a member is mid-push. All three methods must be safe under concurrent invocation. Avoid spawning unbounded thread pools or background tasks; bound any internal executors to a fixed size.
• Failure mode visibility. Surface plugin-side errors (storage outages, retries exhausted, deserialization failures) through the future's exceptional completion rather than logging-and-returning-success. The broker's error path is the only mechanism that distinguishes a successful push from a silent loss.

Security Considerations

The topology description contains user-defined processor names, state-store names, and topic names. This KIP does not treat these as inherently sensitive and does not introduce any client-side redaction. Operators who consider node, store, or topic names sensitive should scope DESCRIBE ACLs on the GROUP resource accordingly: the same ACL that guards StreamsGroupDescribe guards the TopologyDescription field on the response. The UpdateStreamsGroupTopologyDescription path is guarded by READ on the GROUP, identical to the existing heartbeat ACL.

...

The following broker-side sensors metrics are added on GroupCoordinatorMetrics, each exposed as a rate + count meter under the existing group-coordinator-metrics group (JMX type kafka.server:type=group-coordinator-metrics). Each sensor produces both a rate and a cumulative count.

...

Multi-version describe. The describe response surfaces only the topology under the current group's current topologyEpoch. During a rolling topology upgrade, the previous epoch's description may is still stored by the plugin but is no longer reachable via describe. A natural extension is a descriptions[] array on the response, tagged by epoch, allowing operators to view both the previous and the in-flight new topology while the rollout completes.

...