...

Public Interfaces

New Broker Configuration

New Client Configuration

StreamsGroupHeartbeatResponse Change

...

{ "name": "TopologyDescriptionIdTopologyDescriptionRequired", "type": "uuidbool", "versions": "N+",
  "nullableVersions": "N+", "default": "nullfalse",
  "about": "When non-null,True if the client should push its currentbroker's topology description plugin does not have an up-to-date topology description taggedfor withthis thisgroup. idThe viaclient UpdateStreamsGroupTopologyDescription.should Nullsend whenthe notopology pushdescription isvia requestedUpdateStreamsGroupTopologyDescription." }

The broker mints a fresh TopologyDescriptionId on group creation and on every topology epoch bump and persists it on the streams group record. After each heartbeat from a non-STALE_TOPOLOGY member, the broker calls sets this field to true when a topology description plugin is configured and plugin.requiresTopologyPush(requestContext, groupId, groupCreationTimeMs, topologyDescriptionId) with the current id; if the plugin topologyEpoch) returns true, the broker includes that id in the heartbeat response, otherwise the field is omittedwhere requestContext is the context of the heartbeat request.

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": "TopologyDescriptionIdTopologyEpoch", "type": "uuidint32", "versions": "0+",
      "about": "The topologyepoch description id received in the most recent heartbeat response, identifying which topology version this push corresponds toof 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. Populated for source nodes." },
      { "name": "SinkTopic", "type": "string", "versions": "0+", "entityType": "topicName",
        "nullableVersions": "0+", "default": "null",
        "about": "The topic this node writes to. Populated for sink nodes." },
         { "name": "Stores", "type": "[]string", "versions": "0+",
        "about": "The state store names accessed by this node. Populated 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 broker client logs the underlying error at INFO level
• 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": "TopologyDescription", "type": "TopologyDescription", "versions": "N+",
  "nullableVersions": "N+", "default": "null",
  "about": "The topology description for this group. Null if not available — see TopologyDescriptionStatus for the reason." },
{ "name": "TopologyDescriptionStatus", "type": "int8", "versions": "N+", "default": "0",
  "about": "When TopologyDescription is null, the reasonThe status of the topology description for this group: 0=NOT_REQUESTED (client did not set IncludeTopologyDescription), 1=NOT_STORED (no topology description has been recorded for this group), 2=ERROR (the broker failed to fetch the topology description; check broker logs). Ignored when TopologyDescription is non-null, 3=AVAILABLE (a topology description is present in the TopologyDescription field)." }

The TopologyDescription common struct mirrors the struct used by UpdateStreamsGroupTopologyDescriptionRequest (same field names and shape). Because Kafka RPC schemas do not share common structs across message files, the struct is duplicated in StreamsGroupDescribeResponse.json.

...

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

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.Uuid;
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.
 *
 * <p>The broker calls {@link #requiresTopologyPush} on every heartbeat to determine
 * whether the client should send its topology description. Implementations that need
 * to consult an external service should kick off that work asynchronously on the
 * first call and return {@code false} until the result is available.
 *
 * <p>Every method takes an {@link AuthorizableRequestContext} as its first argument.
 *
 * <p>Implementations must be thread-safe. {@link #setTopology} may be called
 * concurrently by multiple group members observing the same
 * {@code TopologyDescriptionIdTopologyDescriptionRequired=true} in the theirsame heartbeat responsescycle; concurrent
 * calls with the same {@code (groupId, topologyDescriptionIdtopologyEpoch)} 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.
     *
     * <p>Called on every successful heartbeat. Not called for members in
     * {@code STALE_TOPOLOGY} status. If this method returns {@code true}, the broker
     * includes thesets {@code topologyDescriptionIdTopologyDescriptionRequired=true} in the heartbeat response.
        *
     * <p>This method should not throw. Failure modes should be handled internally
     * 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.
     *
     * <p>This method is on the heartbeat path and must return quickly. Implementations
     * that need to consult an external service should return {@code false} until the
     * result is available.
     *
     * <p>See the KIP's <em>Plugin Implementation Guidelines</em> section for how
     * implementations should handle in-flight tracking, retries, and topology
     * expiration.
     *
     * @param requestContext the context of the heartbeat request
     * @param groupId the streams group ID
     * @param topologyDescriptionIdgroupCreationTimeMs the broker-minted id identifying the current topology timestamp when the group was created. A value
     *        version for this group; opaque to the plugin
     * @return true if the broker should request a topologyof push{@code from0} the client
     */means "unset". Plugins should treat {@code 0}
    boolean requiresTopologyPush(AuthorizableRequestContext requestContext,
*                            as "incarnation indistinguishable" and should  String groupId, Uuid topologyDescriptionId);

not cross-check
      /**
     * Called when a client sends a topology description for a streams group.
     * This method may be called concurrentlystored byepoch multiplevalues membersagainst ofanother thestored samedescription group;that
     * all calls for the same (groupId, topologyDescriptionId) carry identical data.
     *
     * <p>The returned future completes when the topology hasalso beenhad persisted{@code or
groupCreationTimeMs   == 0}.
     * forwarded.@param CompletetopologyEpoch itthe exceptionallytopology withepoch
     * {@link org.apache.kafka.common.errors.InvalidRequestException} to signal
     * that the payload is semantically invalid, or with@return true if the broker should request a topology push from the client
     */
    boolean * {@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}requiresTopologyPush(AuthorizableRequestContext requestContext,
     * to signal that the description is larger than the plugin is willing to
     * store. Any other exception is mapped to
   String  * {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED}.
     *groupId, long groupCreationTimeMs, int topologyEpoch);


     /**
 @param requestContext the context of* theCalled UpdateStreamsGroupTopologyDescriptionwhen request
a client sends a topology *description @paramfor groupId thea streams group ID.
     * @paramThis topologyDescriptionIdmethod themay idbe thiscalled pushconcurrently isby taggedmultiple with,members asof carriedthe insame thegroup;
     * all calls for the same (groupId, topologyEpoch) heartbeatcarry response that asked for it; opaque to the pluginidentical data.
     * @param description the topology description
     * @return<p>The areturned future that completes when the operation is donetopology has been persisted or
     */
 forwarded. All failures CompletableFuture<Void>must setTopology(AuthorizableRequestContext requestContext,
     be signalled by completing the returned future
     * exceptionally — implementations must not throw synchronously from this method.
     * The broker handles the future's completion exception as follows:
     * String groupId, Uuid topologyDescriptionId,{@link org.apache.kafka.common.errors.InvalidRequestException} maps to
     * {@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.
     StreamsGroupTopologyDescription description);

    /**
     * Called@param whenrequestContext athe groupcontext isof explicitlythe deleted via DeleteGroups. Removes any topologyUpdateStreamsGroupTopologyDescription request
     * description@param storedgroupId forthe thisstreams group.
     * ID
     * <p>The@param returnedgroupCreationTimeMs futurethe completestimestamp when the deletiongroup has been processed.was created
     * If it completes exceptionally, the broker logs the error; the outcome does not @param topologyEpoch the topology epoch
     * @param description the topology description
     * affect@return thea DeleteGroupsfuture responsethat returnedcompletes towhen the caller.operation is done
     */
     * @paramCompletableFuture<Void> setTopology(AuthorizableRequestContext requestContext,
 the context of the DeleteGroups request
     * @param groupId the streams group ID
     * @return a future that completes when the operation is done
     */
   String CompletableFuture<Void> deleteTopology(AuthorizableRequestContext requestContextgroupId, long groupCreationTimeMs, String groupId);

int topologyEpoch,
      /**
     * Called to retrieve the stored topology description for a group. This is invoked
     * by the broker when a client calls StreamsGroupDescribe with
  StreamsGroupTopologyDescription description);

    /**
   {@code IncludeTopologyDescription=true}.
 * Called when a *
group is explicitly deleted via *DeleteGroups. <p>ReturnsRemoves aany futuretopology
 that resolves to the stored* topologydescription descriptionstored for this thegroup.
     * given
 {@code (groupId, topologyDescriptionId)} pair, or* to<p>The {@codereturned null}future ifcompletes no
when the deletion has been *processed.
 topology is stored (e.g. no* pushIf hasit succeededcompletes yetexceptionally, or the storedbroker description
logs     * is tagged with a different id). If the future completes exceptionally, thethe error; the outcome does not
     * pluginaffect signalsthe aDeleteGroups readresponse errorreturned forto thisthe groupcaller.
     *
     * @param requestContext the context of the StreamsGroupDescribeDeleteGroups request
     * @param groupId the streams group ID
     * @param@return topologyDescriptionIda thefuture idthat of the topology versioncompletes when the calleroperation is done
     */
    CompletableFuture<Void> deleteTopology(AuthorizableRequestContext requestContext, String asking about; opaque to the plugingroupId);

    /**
     * @returnCalled ato futureretrieve resolvingthe tostored thetopology storeddescription topologyfor description,a orgroup. nullThis ifis noneinvoked
     */ by the broker when a client calls StreamsGroupDescribe with
    CompletableFuture<StreamsGroupTopologyDescription> * {@code IncludeTopologyDescription=true}.
     *
   getTopology(AuthorizableRequestContext requestContext,
 * <p>Returns a future that resolves to the stored topology description for the
     * given String{@code (groupId, UuidgroupCreationTimeMs, topologyDescriptionId);
}

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

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

public class StreamsGroupTopologyDescription {
    public Collection<Subtopology> subtopologies();
    public Collection<GlobalStore> globalStores();
topologyEpoch)} tuple, or to
     * {@code null} if no topology is stored (e.g. no push has succeeded yet, or the
    public static* classstored Subtopologydescription {
is for a different epoch). If the future publiccompletes
 String id();
   * exceptionally, the plugin signals publica Collection<Node> nodes();
    }
read error for this group.
    public interface*
 Node {
   * @param requestContext the context String name();
  of the StreamsGroupDescribe request
     * Set<String> successors();
    }

@param groupId the streams group ID
     * public static class Source implements Node {@param groupCreationTimeMs the timestamp when the group was created
     * @param topologyEpoch publicthe Set<String> topics();
    }

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

    public static class Sink implements Node {
getTopology(AuthorizableRequestContext requestContext,
         public String topic();
    }

    public staticString classgroupId, GlobalStorelong {
groupCreationTimeMs,        public Source source();
        public Processor processor();
    }
}

Admin Client Interface

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

...

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(); 

        /** Direct successor nodes. */ 
        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 String topic();
    }

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

...

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

   includes a TopologyDescriptionId

   sets TopologyDescriptionRequired in heartbeat responses, in normal operation the RPC is only sent against a broker that has a plugin configured.

   The broker mints a fresh TopologyDescriptionId (a UUID) on group creation and on every topology epoch bump, and persists it on the streams group record as a tagged field.

2. After a successful StreamsGroupHeartbeat, the broker calls requiresTopologyPush on the plugin

   with the current id

   . STALE_TOPOLOGY members are skipped. If the plugin returns true, the broker

   includes the id in

   sets TopologyDescriptionRequired=true on the heartbeat response

   and the client pushes its topology under that id; otherwise the field is omitted

   . 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

   plugin sees only the opaque id and the group identity; it does not see topology epochs or wall-clock timestamps

   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, passing the TopologyDescriptionId from the request unchanged. 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

   reads the current TopologyDescriptionId from group metadata and

   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

   TopologyDescriptionId

   TopologyDescriptionRequired flag from each heartbeat response.

   A non-null id means "push your topology under this id"; a null/missing id means no push is needed.

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

   (carrying the most recently received TopologyDescriptionId)

   when a coordinator is known, the

   id

   flag is

   non-null

   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.

...

NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery; the pending id is preservedflag stays set. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the pending id in place 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 pending id topologyDescriptionRequired flag and log at WARN. The client does not retry on its own; re-attempts, if any, happen when the broker sends a non-null id on re-sets the flag via a subsequent heartbeat. See Plugin Implementation Guidelines for when the plugin should re-solicit.

...

• 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 ids 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 an id once that id has been confirmed too large.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 Avoid concurrent or repetitive pushes: track in-flight pushes per id, back off with an exponential schedule on transient failures, and disable pushes permanently for the id on permanent failures (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 under at the current idepoch.

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.

...

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

...

This KIP bumps StreamsGroupHeartbeatResponse, StreamsGroupDescribeRequest, and StreamsGroupDescribeResponse to the next available version of each RPC and adds the new fields (TopologyDescriptionIdTopologyDescriptionRequired, 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.

...

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

During a rolling upgrade of the Streams application (topology epoch change), the broker mints a new TopologyDescriptionId and skips the plugin entirely for STALE_TOPOLOGY members — neither requiresTopologyPush is called nor is the response flag set. Only members running the new topology reach the plugin and receive the new id. If every active member is stale during the rollout, the current topology remains uncaptured until at least one member heartbeats at the new epoch. 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 running different topology code at the same topology epoch (e.g., a partial deployment), which the broker-minted id alone cannot catch. 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.

...

• 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.Topology descriptions exceeding the configured size limit are rejected; descriptions at the limit are acceptedA 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.

• During a rolling upgrade the feature degrades gracefully on either side: pre-upgrade clients are not asked to push, and pre-upgrade brokers cleanly reject pushes from upgraded clientsA 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).

Rejected Alternatives

Embedding the topology description in the heartbeat

...

{
  "GroupId": "orders-app",
   "TopologyDescriptionIdTopologyEpoch": "8d7e3c2a-4f6b-4d9a-9c1b-7e5f3a8d2c4e"0,
   "TopologyDescription": {
    "Subtopologies": [
      {
        "SubtopologyId": "0",
        "Nodes": [
          {
            "Name": "KSTREAM-SOURCE-0000000000",
            "NodeType": 1,
            "SourceTopics": ["orders"],
            "SinkTopic": null,
            "Stores": [],
            "Successors": ["KSTREAM-FILTER-0000000001"]
          },
          {
            "Name": "KSTREAM-FILTER-0000000001",
            "NodeType": 2,
            "SourceTopics": [],
            "SinkTopic": null,
            "Stores": [],
            "Successors": ["KSTREAM-SINK-0000000002"]
          },
          {
            "Name": "KSTREAM-SINK-0000000002",
            "NodeType": 3,
            "SourceTopics": [],
            "SinkTopic": "valid-orders",
            "Stores": [],
            "Successors": []
          }
        ]
      }
    ],
    "GlobalStores": []
  }
}

...