...

New Client Configuration

...

{
  "apiKey": "TBD",
  "type": "request",
  "listeners": ["broker"],
  "name": "UpdateStreamsGroupTopologyDescriptionRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
	{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
	  "about": "The streams group identifier." },
	{ "name": "MemberId", "type": "string", "versions": "0+",
	  "about": "The ID of the streams group member sending the push." },
	{ "name": "TopologyEpoch", "type": "int32", "versions": "0+",
 	  "about": "The epoch of the topology being described." },
    { "name": "TopologyDescription", "type": "TopologyDescription", "versions": "0+",
      "about": "The topology description." }
  ],
  "commonStructs": [
    { "name": "TopologyDescription", "versions": "0+", "fields": [
      { "name": "Subtopologies", "type": "[]Subtopology", "versions": "0+",
        "about": "The subtopologies that make up this topology." },
      { "name": "GlobalStores", "type": "[]GlobalStore", "versions": "0+",
        "about": "Global state stores used by this topology." }
    ]},
    { "name": "Subtopology", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier, unique within the topology." },
      { "name": "Nodes", "type": "[]TopologyNode", "versions": "0+",
        "about": "The processing nodes in this subtopology." }
    ]},
    { "name": "TopologyNode", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of this node (e.g., KSTREAM-SOURCE-0000000000)." },
      { "name": "NodeType", "type": "int8", "versions": "0+",
        "about": "The type of this node: 1=SOURCE, 2=PROCESSOR, 3=SINK." },
      { "name": "SourceTopics", "type": "[]string", "versions": "0+", "entityType": "topicName",
        "about": "The source topics this node reads from. Defined only for source nodes, may be empty if source topics are dynamically determined." },
      { "name": "SinkTopic", "type": "string", "versions": "0+", "entityType": "topicName",
        "nullableVersions": "0+", "default": "null",
        "about": "The topic this node writes to. Defined only for sink nodes, may be null if sink topic is dynamically determined." },
      { "name": "Stores", "type": "[]string", "versions": "0+",
        "about": "The state store names accessed by this node. Defined only for processor nodes." },
      { "name": "Successors", "type": "[]string", "versions": "0+",
        "about": "The names of successor nodes in the processing graph. Predecessor relationships are reconstructed from this field on the read side." }
    ]},
    { "name": "GlobalStore", "versions": "0+", "fields": [
      { "name": "Source", "type": "TopologyNode", "versions": "0+",
        "about": "The source node providing data to the global store." },
      { "name": "Processor", "type": "TopologyNode", "versions": "0+",
        "about": "The processor node that populates the global store." }
    ]}
  ]
}

...

• GROUP_AUTHORIZATION_FAILED — the client is not authorized

• INVALID_REQUEST — the request is malformed (including an empty MemberId), or the plugin semantically rejected the payload by completing its future with InvalidRequestException

• UNSUPPORTED_VERSION — the coordinator cannot serve this RPC because no topology description plugin is configured
• TOPOLOGY_DESCRIPTION_TOO_LARGE — the plugin rejected the description because it exceeds the size the plugin is willing to store

• TOPOLOGY_DESCRIPTION_UPDATE_FAILED — the plugin failed to process the request for some other reason

  ; the client logs the underlying error at INFO level

• UNKNOWN_MEMBER_ID — the member named in MemberId is no longer in the group

  , or the group itself has been deleted

  ; the client should treat itself as fenced and rejoin

• GROUP_ID_NOT_FOUND — the specified group does not exist
• NOT_COORDINATOR — the broker is not the coordinator for this group
• COORDINATOR_NOT_AVAILABLE — the coordinator is not available
• COORDINATOR_LOAD_IN_PROGRESS — the coordinator is loading

...

 {"name": "IncludeTopologyDescription", "type": "bool", "versions": "N+", "default": "false",
  "about": "Whether to include the full topology description from the topology description plugin in the response." } 

A client that negotiates an older version is handled unchanged by any broker. The flag may only be set when version N or later is negotiated.Clients on older versions never see the flag and behave unchanged.

StreamsGroupDescribeResponse Change

...

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

import org.apache.kafka.common.Configurable;
import java.util.concurrent.CompletableFuture;

/**
 * A broker-side plugin that manages stores, forwards, or exposes topology descriptions for streams groupspushed
 * by Kafka Streams clients.
 *
 * <p>Implementations receivemust topology descriptions pushed by Kafka Streams clientsbe thread-safe. {@link #setTopology} may be called
 * andconcurrently canby store,multiple forward,members orof exposethe themsame howevergroup; theycalls seewith fit. The broker isthe same
 * authoritative on whether a topology is currently stored: it persists a{@code (groupId, topologyEpoch)} carry identical data and must be idempotent.
 * {@code@link StoredTopologyEpoch#deleteTopology} field on the streams-group metadata record and uses it must also be idempotent — it may be called more than once
 * tofor decidethe whethersame to solicit a fresh push on the heartbeat path. Plugins therefore
 * do not need to maintain a per-tuple state machine to answer "do I have this?" —
 * that question has a broker-side answer.
 *
 * <p>Implementations must be thread-safe. {@link #setTopology} may be called
 * concurrently by multiple group members observing
 * {@code TopologyDescriptionRequired=true} in the same heartbeat cycle; concurrent
 * calls with the same {@code (groupId, topologyEpoch)} pair carry identical data
 * and must be treated as idempotent. {@link #deleteTopology} may be called multiple
 * times for the same {@code groupId} if a prior call's bookkeeping write failed; it
 * must also be idempotent when the group has no stored topology.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Called when a client sends a topology description for a streams group.{@code groupId}, including when nothing is stored.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Store the topology description for a streams group.
     *
     * <p>The returned future completes when the topology has been persisted or forwarded.
     * Failures must be signalled by completing the future exceptionally — implementations
     * must not throw synchronously. The completion exception maps to the client-visible
     * error code:
     *
     * <ul>
     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException} — payloads the
     *       plugin will not accept on semantic grounds; reported as {@code INVALID_REQUEST}.</li>
     * This method may be called concurrently by multiple members of the same group;
<li>{@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException} —
     *       descriptions *larger allthan calls for the sameplugin (groupId, topologyEpoch) carry identical data.
     *is willing to store; reported as
     * <p>The returned future completes when the topology has been persisted or{@code TOPOLOGY_DESCRIPTION_TOO_LARGE}.</li>
     * forwarded. All failures<li>Any mustother beexception signalled— bytransient completingbackend thefailure; returnedreported futureas
     * exceptionally — implementations must not throw synchronously from this method.{@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED}.</li>
     * The</ul>
 broker handles the future's completion exception (when present) as follows: *
     *
 The first two are *treated <ul>
as permanent at this topology *epoch and no <li>{@link org.apache.kafka.common.errors.InvalidRequestException} maps tofurther push
     * will be solicited until the epoch {@code INVALID_REQUEST} — use it for payloads the plugin cannot accept onadvances. The third is treated as transient and
     * may be retried.
     */
    CompletableFuture<Void> setTopology(String  semantic grounds. The broker persists this as a permanent-failure
groupId, int topologyEpoch,
               *       {@code LastFailedTopologyEpoch} so subsequent heartbeats do not re-solicit
     *      StreamsGroupTopologyDescription atdescription);

 the same topology epoch.</li>**
     * Remove any <li>{@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
     *       maps to {@code TOPOLOGY_DESCRIPTION_TOO_LARGE} — use it when the description
     * topology description stored for this group. Called when the group is
     * deleted or expires. Failures are logged by the broker but do not propagate to the
     * isuser-visible largerresult thanof thewhatever pluginoperation istriggered willingthe to store. Same permanent-failureremoval.
     */
    CompletableFuture<Void> deleteTopology(String groupId);

 treatment  as above.</li>**
     * Return the <li>Anystored othertopology exceptiondescription maps tofor {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED} and is (groupId, topologyEpoch)}, or
     * {@code null} if the plugin no loggedlonger athas WARN.the The broker treats it as transient and arms an in-memory back-offdata (e.g. backend wipe). If the future
     * completes exceptionally, the broker reports a forread thiserror (groupId, topologyEpoch): the heartbeat path will not re-solicit a freshfor the group.
     */
    CompletableFuture<StreamsGroupTopologyDescription> getTopology(String groupId, push until the back-off window has elapsed. Consecutive transient failures doubleint topologyEpoch);
}

The plugin uses a StreamsGroupTopologyDescription POJO that mirrors org.apache.kafka.streams.TopologyDescription but lives in the org.apache.kafka.coordinator.group.api.streams package; plugin implementations only need to depend on group-coordinator-api. The only difference is that there is no predecessor relation.

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

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

 the back-off, starting atpublic 30static sclass andSubtopology capped{
 at 1 h; a successful push or a
public String id();
   *     public  topology-epoch advance clears the state.</li>Collection<Node> nodes();
    }

	/**
     * </ul>
A processing node in the *
topology. Predecessor nodes can be *inferred @paramfrom groupId the streams group IDsuccessor relation.
     * @param topologyEpoch the topology epoch/
    public interface Node {
     * @param description the topology descriptionString name();
     * @return a future that completes when the operation is doneSet<String> successors();
    }

    public */
static class Source implements CompletableFuture<Void> setTopology(String groupId, int topologyEpoch,Node {
        public Set<String> topics();
    }

    public static class Processor implements Node {
        public        StreamsGroupTopologyDescription descriptionSet<String> stores();

    /**}

    public *static Calledclass whenSink theimplements brokerNode removes{
  a  streams  group.  Removespublic any topology
     * descriptions stored for this group.
     *Optional<String> topic();
    }

    public *static <p>Invokedclass onGlobalStore two{
 paths: when a client deletes the group viapublic {@code DeleteGroups}Source source();
     * (before the grouppublic tombstone is written), and from the broker-internal periodic
     * topology-description cleanup when a streams group becomes empty and all its
     * offsets have expired.
     *
     * <p>The returned future completes when the deletion has been processed.
     * If it completes exceptionally, the broker logs the error; the outcome does not
     * affect the user-visible result of the operation that triggered the removal.
     * The broker may call this method multiple times for the same {@code groupId} if a
     * prior call's bookkeeping write failed — implementations must be idempotent.
     *
     * @param groupId the streams group ID
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> deleteTopology(String groupId);

    /**
     * Called to retrieve the stored topology description for a group. This is invoked
     * by the broker when a client calls StreamsGroupDescribe with
     * {@code IncludeTopologyDescription=true}, but only when the broker-side
     * {@code StoredTopologyEpoch} matches the group's current topology epoch — otherwise
     * the describe path returns {@code NOT_STORED} without invoking this method.
     *
     * <p>Returns a future that resolves to the stored topology description for the
     * given {@code (groupId, topologyEpoch)} pair, or to {@code null} if the plugin has
     * lost its data (e.g. backend wipe). In that case the broker self-heals by clearing
     * {@code StoredTopologyEpoch} so the next heartbeat re-solicits a fresh push.
     *
     * <p>If the future completes exceptionally, the broker signals a read error for this
     * group.
     *
     * @param groupId the streams group ID
     * @param topologyEpoch the topology epoch the caller is asking about
     * @return a future resolving to the stored topology description, or null if none
     */
    CompletableFuture<StreamsGroupTopologyDescription> getTopology(String groupId, int topologyEpoch);
}

The plugin uses a StreamsGroupTopologyDescription POJO that mirrors org.apache.kafka.streams.TopologyDescription but lives in the org.apache.kafka.coordinator.group.api.streams package; plugin implementations only need to depend on group-coordinator-api. The only difference is that there is no predecessor relation.

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):

 { "name": "StoredTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 3,
  "default": -1, "type": "int32",
  "about": "The topology epoch whose description is currently stored in the topology description plugin, or -1 if none is stored." },
{ "name": "LastFailedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 4,
  "default": -1, "type": "int32",
  "about": "The topology epoch whose description push the plugin permanently rejected (TooLarge / InvalidRequest), or -1 if none. Heartbeat-path solicitation is suppressed while this equals the current topology epoch, to avoid hot-looping." }

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):

 { "name": "StoredTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 3,
  "default": -1, "type": "int32",
  "about": "The topology epoch whose description is currently stored in the topology description plugin, or -1 if none is stored." },
{ "name": "LastFailedTopologyEpoch", "versions": "0+", "taggedVersions": "0+", "tag": 4,
  "default": -1, "type": "int32",
  "about": "The topology epoch whose description push the plugin permanently rejected (TooLarge / InvalidRequest), or -1 if none. Heartbeat-path solicitation is suppressed while this equals the current topology epoch, to avoid hot-looping." }

The two fields together drive the broker-side gating decision on the The two fields together drive the broker-side gating decision on the heartbeat and describe paths. Both are tagged fields, so older brokers that decode the record before deserializing the tags simply see the defaults. Streams groups that existed before this KIP land carry both fields as -1; on the first post-upgrade heartbeat the broker solicits a push, and the regular setTopology path persists the new value.

...

package org.apache.kafka.clients.admin;

public class StreamsGroupTopologyDescription {
    public Collection<Subtopology> subtopologies();
    public Collection<GlobalStore> globalStores();package org.apache.kafka.clients.admin;

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

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

    public interface Node {
        String name();

        /** Direct predecessor nodes. */
        Set<String> predecessors(); 

    public static class  /** Direct successor nodes. */ 
        Set<String> successorsSubtopology {
        public String id();
        public Collection<Node> nodes();
    }

    public static class Source implementsinterface Node {
        publicString Set<String> topicsname();

    }

    public/** staticDirect classpredecessor Processor implements Node {
        publicnodes. */
        Set<String> storespredecessors(); 

    }

    public static class Sink implements Node {/** Direct successor nodes. */ 
        publicSet<String> Optional<String> topicsuccessors();
      }

    public static class Source GlobalStoreimplements Node {
        public SourceSet<String> sourcetopics();
    }

    public static class Processor processor(); implements Node {
    }
}

Command-Line Tool

kafka-streams-groups.sh gains a new --topology sub-action under --describe, parallel to --members, --offsets, and --state:

...

...

...

...

The output mirrors the format produced by Topology#describe() in the Kafka Streams API:

...

 public Set<String> stores();
   

...

 }

    public 

...

static class Sink 

...

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

    public 

...

static class GlobalStore {
      

...

  public Source source();
       

...

 public Processor processor();
   

...

 }
}

Command-Line Tool

kafka-streams-groups.sh gains a new --topology sub-action under --describe, parallel to --members, --offsets, and --state:

kafka-streams-groups.sh --bootstrap-server <broker> --describe --topology --group <group-id>

The output mirrors the format produced by Topology#describe() in the Kafka Streams API:

Topologies:
   Sub-topology: 0
     Source:  KSTREAM-SOURCE-0000000000 (topics: [input-topic])
       --> my-processor
     Processor: my-processor (stores: [my-store])
       <-- KSTREAM-SOURCE-0000000000
       --> KSTREAM-SINK-0000000002
     Sink: KSTREAM-SINK-0000000002 (topic: output-topic)
       <-- my-processor

Global stores, if present, are printed in a trailing Global Stores: block. The command calls the admin client with includeTopologyDescription(true).

When the coordinator returns no topology description, the command picks its output from the TopologyDescriptionStatus on the response:

• NOT_STORED → "No topology description has been recorded for group '<group-id>'."
• ERROR → "The broker failed to retrieve the topology description for group '<group-id>' (check broker logs)."
• AVAILABLE → normal pretty-printed topology.

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. The plugin is instantiated at broker startup if group.streams.topology.description.plugin.class is configured. A broker without a plugin returns UNSUPPORTED_VERSION for UpdateStreamsGroupTopologyDescription and never sets TopologyDescriptionRequired, so the RPC is only sent against a plugin-configured broker.

2. After a successful StreamsGroupHeartbeat, the broker decides whether to set TopologyDescriptionRequired=true purely from the group's persisted state — no plugin RPC is involved. Members with STALE_TOPOLOGY status are skipped. For all other members the broker sets the flag iff StoredTopologyEpoch != currentTopologyEpoch AND LastFailedTopologyEpoch != currentTopologyEpoch AND no per-group back-off is in its window. The back-off is in-memory state (keyed by groupId, carrying topologyEpoch + nextAttemptMs) that arms or extends every time the flag is set and additionally on a transient setTopology failure; consecutive arms double the window from 30 s up to 1 h. It clears on a successful push, on a permanent failure (where LastFailedTopologyEpoch ratchets), and implicitly on any topology-epoch advance. The same mechanism covers unresponsive plugins and clients that never push (for example, with topology.description.push.enabled=false).

3. On UpdateStreamsGroupTopologyDescription, the broker checks the READ ACL on the group and that a plugin is configured, then validates the MemberId: an empty MemberId is rejected with INVALID_REQUEST, a non-existing streams group with GROUP_ID_NOT_FOUND, and a MemberId not matching any current member with UNKNOWN_MEMBER_ID. The broker enforces no size limit; the plugin decides what it is willing to store. The broker then calls setTopology on the plugin. On success it writes a metadata record setting StoredTopologyEpoch = pushedEpoch and the response carries NONE. On InvalidRequestException or TopologyDescriptionTooLargeException it writes LastFailedTopologyEpoch = pushedEpoch so subsequent heartbeats at the same 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 metadata-record write fails, the next heartbeat sees the drift, re-solicits, and the idempotent re-push closes the gap.

4. On DeleteGroups, the broker calls deleteTopology on the plugin before writing the group tombstone, for each requested streams group with StoredTopologyEpoch != -1. Plugin failures are logged but do not affect the deletion response; the group is tombstoned regardless. This ordering matches the natural-expiration cleanup below.

5. 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). With no plugin configured the broker returns NOT_STORED. A getTopology call returning null (plugin-side data loss) surfaces as NOT_STORED and is logged at WARN; subsequent describes keep returning NOT_STORED until the topology epoch advances or an operator clears plugin state.
6. When a plugin is configured, the broker runs a periodic topology-description cleanup 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. 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 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.
2. The Streams client records the TopologyDescriptionRequired flag from each heartbeat response.
3. On each consumer background-thread poll, the client 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. The push runs on the consumer background thread and never blocks user-facing Kafka Streams APIs; the push is best-effort.
4. Completion handling on the push response is keyed on the error code. NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery and leave the flag set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set for retry on the next poll. 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 flag and log at WARN; the client does not retry on its own, and a re-attempt happens only when the broker re-sets the flag via a subsequent heartbeat. 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 per-group back-off (30 s → 1 h, exponential) throttles re-solicitation.

• Return null from getTopology when the plugin has lost the description; the broker reports NOT_STORED on the describe response. Exceptions are reserved for transient backend failures and surface as ERROR.

• 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.

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.

Metrics

The following broker-side metrics are added 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.

...

Global stores, if present, are printed in a trailing Global Stores: block. The command calls the admin client with includeTopologyDescription(true).

When the coordinator returns no topology description, the command picks its output from the TopologyDescriptionStatus on the response:

• NOT_STORED → "No topology description has been recorded for group '<group-id>'."
• ERROR → "The broker failed to retrieve the topology description for group '<group-id>' (check broker logs)."
• AVAILABLE → normal pretty-printed topology.

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 Removed

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. When a plugin is configured, the broker runs a periodic topology-description cleanup 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. 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 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.
2. The Streams client records the TopologyDescriptionRequired flag from each heartbeat response.
3. On each consumer background-thread poll, the client 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. The push runs on the consumer background thread and never blocks user-facing Kafka Streams APIs; the push is best-effort.
4. Completion handling on the push response is keyed on the error code. NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery and leave the flag set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set for retry on the next poll. 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 flag and log at WARN; the client does not retry on its own, and a re-attempt happens only when the broker re-sets the flag via a subsequent heartbeat. 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.

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.

Metrics

The following broker-side metrics are added 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.

No client-side metrics are introduced.

Compatibility, Deprecation, and Migration Plan

This KIP bumps StreamsGroupHeartbeatResponse, StreamsGroupDescribeRequest, and StreamsGroupDescribeResponse to the next available version of each RPC and adds the new fields (TopologyDescriptionRequired, IncludeTopologyDescription, TopologyDescription, TopologyDescriptionStatus) at that version. Pre-upgrade clients negotiate an older version and never see the new fields, so all three changes are wire-compatible.

The new RPC uses a new API key and is only sent by clients that understand the feature.

Without a configured plugin, no flags are set and no topology descriptions are sent. There is no behavioral change for existing deployments.

Rolling Upgrades

During a rolling upgrade of brokers, some brokers may have the plugin configured and some may not. The TopologyDescriptionRequired flag is only set by plugin-equipped coordinators; a client whose current coordinator lacks the plugin never sees the flag. If the coordinator migrates mid-push to a plugin-less broker, the client receives UNSUPPORTED_VERSION, clears the topologyDescriptionRequired flag, and does not retry. The flag is set again only if a future heartbeat response from a plugin-equipped coordinator includes TopologyDescriptionRequired=true.

During a rolling upgrade of the Streams application (topology epoch change), the broker skips the heartbeat-path gating for STALE_TOPOLOGY members — the response flag is not set, regardless of the persisted StoredTopologyEpoch. Only members running the new topology reach the gating comparison. If every active member is stale during the rollout, the current-epoch topology remains uncaptured until at least one member heartbeats with the new epoch.

During this transition the assignment topology (advanced synchronously when a new-epoch member heartbeats) and the description topology (advanced asynchronously via the plugin) can briefly disagree: StreamsGroupDescribe may report the new topologyEpoch while TopologyDescription is still null with status NOT_STORED until the first push for the new epoch succeeds. The two reconverge once any member at the new epoch pushes its description.

Future Work

Hash-based mismatch detection. A future enhancement could introduce a topology hash to detect clients on different topology descriptions reporting the same topology epoch.

Multi-version describe. The describe response surfaces only the topology under the group's current topologyEpoch. During a rolling topology upgrade, the previous epoch's description 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.

Node grouping. A nodeGroup field on TopologyNode for UI grouping, requiring an extension to the public TopologyDescription.Node interface in Kafka Streams — best addressed in a follow-up KIP.

Test Plan

Integration Tests

• A Streams client pushes its topology description after the broker requests it, the configured plugin receives the description, and the group's StoredTopologyEpoch is updated to the pushed epoch.
• A broker without a topology description plugin never asks the client to push and rejects pushes outright.
• Authorization is enforced on the new RPC.
• A plugin completing the setTopology future with TopologyDescriptionTooLargeException causes the broker to return TOPOLOGY_DESCRIPTION_TOO_LARGE, persists LastFailedTopologyEpoch = pushedEpoch, and subsequent heartbeats at the same epoch do not re-solicit.
• A setTopology plugin success followed by an injected metadata-record commit failure recovers on the next heartbeat: the broker re-solicits, the client re-pushes, the plugin's idempotent setTopology is invoked a second time, and StoredTopologyEpoch ends up correctly set.
• Requesting a topology description via describe returns it for groups whose StoredTopologyEpoch matches the current epoch, and surfaces NOT_STORED (no description stored or epoch mismatch) or ERROR (plugin exception) otherwise — without turning the describe itself into an error.
• A getTopology returning null while the broker believed the description was stored triggers a fire-and-forget StoredTopologyEpoch = -1 write; the next heartbeat re-solicits.
• An explicit DeleteGroups for a streams group with StoredTopologyEpoch != -1 calls plugin.deleteTopology before tombstoning the group; the group is tombstoned even if the plugin call fails.
• A push from a member that no longer belongs to the group (or whose group has been deleted) is rejected with UNKNOWN_MEMBER_ID; the client clears its push flag and rejoins via the existing heartbeat-fence path. A push with an empty MemberId is rejected with INVALID_REQUEST.
• The CLI prints the topology when it is available and reports a clear, distinct message for each non-available outcome, with exit code 0 only when the topology is actually returned.

System Tests

No client-side metrics are introduced.

Compatibility, Deprecation, and Migration Plan

This KIP bumps StreamsGroupHeartbeatResponse, StreamsGroupDescribeRequest, and StreamsGroupDescribeResponse to the next available version of each RPC and adds the new fields (TopologyDescriptionRequired, IncludeTopologyDescription, TopologyDescription, TopologyDescriptionStatus) at that version. Pre-upgrade clients negotiate an older version and never see the new fields, so all three changes are wire-compatible.

The new RPC uses a new API key and is only sent by clients that understand the feature.

Without a configured plugin, no flags are set and no topology descriptions are sent. There is no behavioral change for existing deployments.

Rolling Upgrades

During a rolling upgrade of brokers, some brokers may have the plugin configured and some may not. The TopologyDescriptionRequired flag is only set by plugin-equipped coordinators; a client whose current coordinator lacks the plugin never sees the flag. If the coordinator migrates mid-push to a plugin-less broker, the client receives UNSUPPORTED_VERSION, clears the topologyDescriptionRequired flag, and does not retry. The flag is set again only if a future heartbeat response from a plugin-equipped coordinator includes TopologyDescriptionRequired=true.

During a rolling upgrade of the Streams application (topology epoch change), the broker skips the heartbeat-path gating for STALE_TOPOLOGY members — the response flag is not set, regardless of the persisted StoredTopologyEpoch. Only members running the new topology reach the gating comparison. If every active member is stale during the rollout, the current-epoch topology remains uncaptured until at least one member heartbeats with the new epoch.

During this transition the assignment topology (advanced synchronously when a new-epoch member heartbeats) and the description topology (advanced asynchronously via the plugin) can briefly disagree: StreamsGroupDescribe may report the new topologyEpoch while TopologyDescription is still null with status NOT_STORED until the first push for the new epoch succeeds. The two reconverge once any member at the new epoch pushes its description.

Future Work

Hash-based mismatch detection. A future enhancement could introduce a topology hash to detect clients on different topology descriptions reporting the same topology epoch.

Multi-version describe. The describe response surfaces only the topology under the group's current topologyEpoch. During a rolling topology upgrade, the previous epoch's description 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.

Node grouping. A nodeGroup field on TopologyNode for UI grouping, requiring an extension to the public TopologyDescription.Node interface in Kafka Streams — best addressed in a follow-up KIP.

Test Plan

Integration Tests

Broker:

• UpdateStreamsGroupTopologyDescriptionRequestTest (new) — push happy path; permanent and transient plugin failures; heartbeat-flag gating; member/group/MemberId fencing; explicit DeleteGroups ordering.
• UpdateStreamsGroupTopologyDescriptionNoPluginRequestTest (new) — broker without plugin: pushes rejected, flag never set, describe returns NOT_STORED.
• AuthorizerIntegrationTest — READ ACL on the GROUP resource for the new RPC.

Streams client:

• TopologyDescriptionPluginIntegrationTest (new) — end-to-end push against an in-memory plugin; describe status mapping; topology.description.push.enabled=false opt-out.

CLI:

• TopologyDescriptionFormatterTest (new) — each TopologyDescriptionStatus, the pretty-print format, and the exit-code mapping.
• StreamsGroupCommandTest — the --topology sub-action against a populated group.

System Tests

New ducktape suite streams_topology_description_plugin_test.py driving the Java harness TopologyDescriptionPluginSystemTest:

• StreamsTopologyDescriptionPluginTest — CLI describe of a running Kafka Streams app; client opt-out; periodic-cleanup-after-retention. The last scenario lives here because realistic offsets.retention.minutes is minutes-to-hours; overriding it low enough at the integration layer would affect unrelated tests on the shared embedded cluster.
• StreamsTopologyDescriptionPluginNoPluginTest — broker without plugin reports NOT_STORED and never asks the client to push
• A running Kafka Streams application pushes its topology and an operator can retrieve and pretty-print it via the CLI end-to-end.

• Disabling the feature on the client (topology.description.push.enabled=false) stops it from sending topology descriptions altogether; describe returns NOT_STORED.

• A broker without the plugin configured cleanly reports NOT_STORED on describe and never asks the client to push (the broker-side half of the rolling-upgrade matrix; the client-side half is intrinsic to wire-version negotiation).
• Configure a short offsets.retention.minutes and offsets.retention.check.interval.ms, run a streams app that pushes its topology, stop it, wait for the offsets to expire and the cleanup timer to fire, and verify plugin.deleteTopology was called and the group was tombstoned. Run as a system test because the realistic timing is minutes-to-hours and overriding offsets.retention.minutes low enough at the integration-test layer would impact the shared embedded-cluster's offset semantics for unrelated tests.

Rejected Alternatives

Embedding the topology description in the heartbeat

...