...

The broker sets this field to to true when a topology description plugin is configured and plugin.requiresTopologyPush(requestContext, groupId, groupCreationTimeMs, topologyEpoch) returns true, where requestContext is the context of the heartbeat requestthe broker expects the client to send it's current version of the topology description.

New RPC: UpdateStreamsGroupTopologyDescription (API Key TBD)

...

{
  "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": "TopologyEpochMemberId", "type": "int32string", "versions": "0+",
 	     "about": "The epochID of the topology being described streams group member sending the push." },
    	{ "name": "TopologyDescriptionTopologyEpoch", "type": "TopologyDescriptionint32", "versions": "0+",
 	     "about": "The epoch of the topology being descriptiondescribed." },
   ],
  "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, 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_NOTGROUP_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

...

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

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

/**
 * A broker-side plugin that manages topology descriptions for streams groups.
 *
 * <p>Implementations receive topology descriptions pushed by Kafka Streams clients
 * and can store, forward, or expose them however they see fit. The broker is
 *
 *authoritative <p>Theon brokerwhether callsa {@linktopology #requiresTopologyPush}is oncurrently everystored: heartbeatit topersists determinea
 * whether{@code theStoredTopologyEpoch} clientfield shouldon sendthe itsstreams-group topologymetadata description.record Implementationsand thatuses needit
 * to decide consultwhether anto externalsolicit servicea shouldfresh kickpush offon thatthe workheartbeat asynchronouslypath. onPlugins thetherefore
 * firstdo callnot andneed returnto {@codemaintain false} until the result is available.
 *
 * <p>Every method takes an {@link AuthorizableRequestContext} as its first argumenta 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.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Returns whether the broker should request a topology push from the client.
      {@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 {

    /**
     * <p>CalledCalled when ona everyclient successfulsends heartbeat.a Nottopology calleddescription for members ina streams group.
     * {@code STALE_TOPOLOGY} status. If this method returns {@code true}, the brokerThis method may be called concurrently by multiple members of the same group;
     * sets {@code TopologyDescriptionRequired=true} in the heartbeat response.
      all calls for the same (groupId, topologyEpoch) carry identical data.
     *
     * <p>This<p>The methodreturned shouldfuture notcompletes throw.when Failurethe modestopology shouldhas bebeen handledpersisted internallyor
     * and converted to a return value of {@code false}. The broker defensively
     * catches any exception and treats it as {@code false}, logging at WARN; plugins
     * should not rely on this backstop.forwarded. All failures must be signalled by completing the returned future
     * exceptionally — implementations must not throw synchronously from this method.
     * The broker handles the future's completion exception (when present) as follows:
     *
     * <p>This<ul>
 method is on the heartbeat* path and must return quickly. Implementations<li>{@link org.apache.kafka.common.errors.InvalidRequestException} maps to
     * that need to consult an external service should return { {@code falseINVALID_REQUEST} until— the
use it for payloads the *plugin resultcannot isaccept available.on
     *
      * <p>Seesemantic the KIP's <em>Plugin Implementation Guidelines</em> section for howgrounds. The broker persists this as a permanent-failure
     * implementations should handle in-flight tracking, retries, and topology
     * expiration.{@code LastFailedTopologyEpoch} so subsequent heartbeats do not re-solicit
     *
     * @param requestContextat the contextsame of the heartbeat requesttopology epoch.</li>
     * @param groupId the streams group ID<li>{@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
     * @param groupCreationTimeMs the timestamp when the groupmaps wasto created. A value
     *{@code TOPOLOGY_DESCRIPTION_TOO_LARGE} — use it when the description
     *       is larger than the plugin is willing to store. Same permanent-failure
     * of {@code 0} means "unset". Plugins shouldtreatment treat {@code 0}as above.</li>
     *   <li>Any other exception maps to      {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED} and is
     *       logged at WARN. The broker treats it as "incarnationtransient indistinguishable"and andarms shouldan notin-memory crossback-checkoff
     *       for this (groupId, topologyEpoch): the heartbeat path will not re-solicit a fresh
     *     stored epoch valuespush againstuntil anotherthe storedback-off descriptionwindow that
has elapsed. Consecutive transient failures *double
     *       the back-off, starting at 30 s and capped at 1 h; a successful push or a
 also  had {@code groupCreationTimeMs* == 0}.
     *topology-epoch @paramadvance topologyEpochclears the topology epochstate.</li>
     * </ul>
  @return true if the*
 broker should request a topology* push@param fromgroupId the streams clientgroup ID
     */
 @param topologyEpoch the boolean requiresTopologyPush(AuthorizableRequestContext requestContext,topology epoch
     * @param description the topology description
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> setTopology(String groupId, longint groupCreationTimeMstopologyEpoch,
  int topologyEpoch);


    /**
     * Called when a client sends a topology description for a streams group.
     * This method may be called concurrently by multiple members of the sameStreamsGroupTopologyDescription groupdescription);

    /**
 * all calls for the* sameCalled (groupId, topologyEpoch) carry identical data.
     *when the broker removes a streams group. Removes any topology
     * <p>Thedescriptions returnedstored futurefor completesthis whengroup.
 the topology has been persisted or*
     * forwarded.<p>Invoked Allon failurestwo mustpaths: bewhen signalleda byclient completingdeletes the returned future group via {@code DeleteGroups}
     * exceptionally(before —the implementationsgroup musttombstone notis throwwritten), synchronouslyand from the this method.broker-internal periodic
     * The broker handles the future's completion exception as follows:topology-description cleanup when a streams group becomes empty and all its
     * offsets {@linkhave org.apache.kafka.common.errors.InvalidRequestException} maps toexpired.
     *
     * {@code INVALID_REQUEST}; {@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
     * maps to {@code TOPOLOGY_DESCRIPTION_TOO_LARGE}; any other exception maps to
     * {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED} and is logged at WARN.<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 @paramthe requestContextoperation thethat context oftriggered the UpdateStreamsGroupTopologyDescription requestremoval.
     * @paramThe groupIdbroker the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created
     * @param topologyEpoch the topology epochmay call this method multiple times for the same {@code groupId} if a
     * prior call's bookkeeping write failed — implementations must be idempotent.
     *
     * @param descriptiongroupId the topologystreams group descriptionID
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> setTopologydeleteTopology(AuthorizableRequestContext requestContext,String groupId);

    /**
     * Called to retrieve the stored topology description for a group. This is invoked
     * by the broker when a client calls StreamsGroupDescribe with
    String groupId,* long groupCreationTimeMs{@code IncludeTopologyDescription=true}, intbut topologyEpoch,
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.
     StreamsGroupTopologyDescription description);

    /**
     * Called<p>Returns when a groupfuture isthat explicitlyresolves deletedto viathe DeleteGroups. Removes any topology
     *stored topology description stored for this group.the
     *
 given {@code (groupId, topologyEpoch)} *pair, <p>Theor returnedto future{@code completesnull} whenif the deletionplugin has been processed.
     * Iflost itits completes exceptionally, the broker logs the error; the outcome does notdata (e.g. backend wipe). In that case the broker self-heals by clearing
     * affect{@code StoredTopologyEpoch} so the DeleteGroupsnext responseheartbeat returnedre-solicits toa thefresh callerpush.
     *
     * @param<p>If requestContextthe thefuture contextcompletes ofexceptionally, the DeleteGroupsbroker request
signals a read error for this
 * @param groupId the streams* group ID.
     *
 @return  a future that* completes@param whengroupId the operationstreams isgroup doneID
     */
    CompletableFuture<Void> deleteTopology(AuthorizableRequestContext requestContext, String groupId);

 @param topologyEpoch the topology epoch the caller is asking about
     /**
 @return a future  *resolving Called to retrieve the stored topology description, foror anull group. This is invokedif none
     */
 by the broker whenCompletableFuture<StreamsGroupTopologyDescription> a client calls StreamsGroupDescribe with
     * {@code IncludeTopologyDescription=true}.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();
   * <p>Returnspublic a future that resolves to the stored topology description for theCollection<GlobalStore> globalStores();

    public static class Subtopology {
     * given {@code (groupId, groupCreationTimeMs, topologyEpoch)} tuple, or to
public String id();
        * {@code null} if no topology is stored (e.g. no push has succeeded yet, or the
     * stored description is for a different epoch). If the future completespublic Collection<Node> nodes();
    }

	/**
     * A processing node in the topology. Predecessor nodes can be inferred from successor relation.
     */
  exceptionally,  thepublic plugininterface signalsNode a{
 read error for this group.
     *String name();
     * @param requestContext the context of the StreamsGroupDescribe requestSet<String> successors();
    }

    public *static @paramclass groupIdSource theimplements streamsNode group ID{
     * @param groupCreationTimeMs thepublic timestamp when the group was createdSet<String> topics();
    }

    public *static @paramclass topologyEpochProcessor theimplements topologyNode epoch{
 the caller is asking about
   public  * @return a future resolving to the stored topology description, or null if none
     */
    CompletableFuture<StreamsGroupTopologyDescription>Set<String> stores();
    }

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

    public static class  getTopology(AuthorizableRequestContext requestContext,GlobalStore {
        public Source source();
        public  String groupId, long groupCreationTimeMs, int topologyEpoch);
}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):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();
    }
}

Admin Client Interface

DescribeStreamsGroupsOptions gains an includeTopologyDescription(boolean) setter. When set to true, the admin client sets IncludeTopologyDescription on the StreamsGroupDescribeRequest (at the new version) and the coordinator consults the plugin.

StreamsGroupDescription gains two accessors:

 { "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 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.

Admin Client Interface

DescribeStreamsGroupsOptions gains an includeTopologyDescription(boolean) setter. When set to true, the admin client sets IncludeTopologyDescription on the StreamsGroupDescribeRequest (at the new version) and the coordinator consults the plugin.

StreamsGroupDescription gains two accessors:

• Optional<StreamsGroupTopologyDescription> topologyDescription() — empty unless the field was requested and the plugin returned a description.
• StreamsGroupTopologyDescriptionStatus topologyDescriptionStatus() — a
• Optional<StreamsGroupTopologyDescription> topologyDescription() — empty unless the field was requested and the plugin returned a description.
• StreamsGroupTopologyDescriptionStatus topologyDescriptionStatus() — a new enum { AVAILABLE, NOT_REQUESTED, NOT_STORED, ERROR }. AVAILABLE is reported when TopologyDescription is non-null; the remaining values mirror the wire-level TopologyDescriptionStatus int8.

...

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

Broker Side

1. The plugin is instantiated at broker startup if group.streams.topology.description.plugin.class is configured. A broker without a plugin does not advertise UpdateStreamsGroupTopologyDescription in its ApiVersions response and returns UNSUPPORTED_VERSION for any such request. Because a plugin-less broker never sets TopologyDescriptionRequired in heartbeat responses, in normal operation the RPC is only sent against a broker that has a plugin configured.

2. After a successful StreamsGroupHeartbeat, the broker calls requiresTopologyPush on the plugin. STALE_TOPOLOGY members are skipped. If the plugin returns true, the broker sets TopologyDescriptionRequired=true on the heartbeat response. Per the plugin contract requiresTopologyPush should not throw; the broker defensively catches any exception, logs at WARN, and treats the call as false. Beyond the STALE_TOPOLOGY skip the broker does no other gating — deduplication, timeouts, and back-off are the plugin's responsibility (see Plugin Implementation Guidelines). The groupCreationTimeMs parameter identifies the specific incarnation of a group; a recreated group with the same groupId gets a fresh creation timestamp. It is persisted on the group record as a tagged field shared with KIP-1282. Groups created before KIP-1282 is implemented carry the sentinel value groupCreationTimeMs = 0.

3. On UpdateStreamsGroupTopologyDescription, the broker checks the READ ACL on the group and that a plugin is configured. 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 response carries NONE; InvalidRequestException from the plugin maps to INVALID_REQUEST, TopologyDescriptionTooLargeException maps to TOPOLOGY_DESCRIPTION_TOO_LARGE, any other exception maps to TOPOLOGY_DESCRIPTION_UPDATE_FAILED, and all three are logged at WARN.
4. On DeleteGroups, the broker calls deleteTopology on the plugin only after the group delete has returned no error for that group. deleteTopology failures are logged but do not affect the deletion response. For groups that expire naturally (all members leave), deleteTopology is not called — the plugin expires the topology via a wall-clock TTL (see Plugin Implementation Guidelines).

5. On StreamsGroupDescribe with IncludeTopologyDescription=true, the broker calls getTopology on the plugin for each group after assembling the rest of the response, passing the group's current (groupCreationTimeMs, topologyEpoch) tuple. 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 (the broker does not distinguish "no plugin" from "plugin has nothing stored" on the wire).

Client Side

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

2. 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. When topology.description.push.enabled=false, no description is stored and the feature is disabled on this client.

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, and no prior request is in flight. 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. The push is best-effort: a failure to send the topology description does not prevent the Streams application from running.

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

NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery; the flag stays set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set for retry on the next poll.

All other errors (TOPOLOGY_DESCRIPTION_TOO_LARGE, TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the topologyDescriptionRequired flag and log at WARN. The client does not retry on its own; re-attempts, if any, happen when the broker re-sets the flag via a subsequent heartbeat. See Plugin Implementation Guidelines for when the plugin should re-solicit.

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

Plugin Implementation Guidelines

Broker-side gating around the plugin is minimal. The plugin can expect requiresTopologyPush to be called on a regular cadence following each group's heartbeat interval.

A correct plugin implementation should:

• Implement requiresTopologyPush as a non-blocking, efficient call, since it is invoked frequently.
• Expire and clean up stored topology descriptions for inactive groups and superseded topology epochs internally. Incoming requiresTopologyPush calls can be used as an implicit keep-alive.
• Decide and enforce a maximum stored description size. Reject pushes that exceed it by completing the setTopology future with TopologyDescriptionTooLargeException, and stop returning requiresTopologyPush=true for the same (groupId, topologyEpoch) pair once that pair has been confirmed too large.
• Throttle re-solicitation without relying on the heartbeat interval. For each (groupId, groupCreationTimeMs, topologyEpoch) tuple the plugin should track (i) whether a push is currently in flight and (ii) the time of the last requiresTopologyPush=true. While a push is in flight requiresTopologyPush returns false. Once the push has completed successfully it returns false permanently for that tuple. On a transient setTopology failure the plugin arranges for requiresTopologyPush to return true again after a self-driven back-off (e.g. exponential, starting at 1s) — this is the client re-solicitation mechanism, and it is the only retry pathway: the wire-level TOPOLOGY_DESCRIPTION_UPDATE_FAILED is terminal from the client's perspective and pairs with this plugin-side re-solicitation. On permanent failure (TOPOLOGY_DESCRIPTION_TOO_LARGE or plugin-semantic INVALID_REQUEST) the plugin returns false permanently for the tuple and logs.
• Not request repeated pushes from stable groups that have already pushed their topology at the current epoch.

See the appendix for an example state machine for managing the lifecycle of a topology description.

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

, 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

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

Client Side

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

2. 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. When topology.description.push.enabled=false, no description is stored and the feature is disabled on this client.

   
   

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. 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. The push is best-effort: a failure to send the topology description does not prevent the Streams application from running.

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

NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery; the flag stays set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set for retry on the next poll.

All other errors (TOPOLOGY_DESCRIPTION_TOO_LARGE, TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the topologyDescriptionRequired flag and log at WARN. The client does not retry on its own; re-attempts, if any, happen when the broker re-sets the flag via a subsequent heartbeat. See Plugin Implementation Guidelines for when the plugin should re-solicit.

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.

Metrics

The following broker-side sensors are added on GroupCoordinatorMetrics, each exposed as a rate + count meter under the existing coordinator metrics group:

No client-side metrics are introduced.This KIP does not introduce any new broker-side or client-side metrics. It is the responsibility of the plugin to define metrics as necessary.

Compatibility, Deprecation, and Migration Plan

...

During a rolling upgrade of the Streams application (topology epoch change), the broker skips the plugin entirely heartbeat-path gating for STALE_TOPOLOGY members — neither requiresTopologyPush is called nor is the response flag is not set, regardless of the persisted StoredTopologyEpoch. Only members running the new topology reach the plugingating 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 current (groupCreationTimeMs, topologyEpoch) tuple. During a rolling topology upgrade, the previous epoch's description may 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

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 current topologyEpoch. During a rolling topology upgrade, the previous epoch's description may 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
• A Streams client pushes its topology description after the broker requests it, and the configured plugin receives the description.
• 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, and the client clears the topologyDescriptionRequired flag without retrying.
• Permanent plugin-side validation failures stop the client from retrying; transient failures and throttling delay subsequent attempts.
• Requesting a topology description via describe returns it for groups that have pushed, and surfaces the appropriate non-available status (no description stored, or read error) otherwise — without turning the describe itself into an error.
• 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.

...

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

  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

...

KIP-714 supports compression (ZStd, LZ4, GZip, Snappy) for telemetry payloads because metrics are pushed repeatedly at high frequency. Topology descriptions are pushed infrequently (only on topology epoch changes) and the wire payload is expected to fit comfortably for typical applications. The Kafka protocol's flexible versions already provide efficient serialization. Adding compression would add complexity without meaningful benefit for this use case.

Appendix: Example plugin lifecycle

Image Removed

Appendix: Example RPC payload

For illustration, consider a Kafka Streams application that reads from orders, filters out null values, and writes the remaining records to valid-orders:

...