...

Discussion thread: here 

JIRA: TBD

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

New Client Configuration

...

{ "name": "TopologyDescriptionRequired", "type": "bool", "versions": "N+", "ignorable": true, "default": "false",
  "about": "True if the client should send the topology description via UpdateStreamsGroupTopologyDescriptionStreamsGroupTopologyDescriptionUpdate." }

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:

...

StreamsGroupTopologyDescriptionUpdate (API Key TBD)

A new RPC is introduced for setting the topology description for a streams group. Like StreamsGroupHeartbeat, the request is sent to the group coordinator for the group.

...

{
  "apiKey": "TBD",
  "type": "request",
  "listeners": ["broker"],
  "name": "UpdateStreamsGroupTopologyDescriptionRequestStreamsGroupTopologyDescriptionUpdateRequest",
  "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 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." }
    ]}
  ]
}

...

{
  "apiKey": "TBD",
  "type": "response",
  "name": "UpdateStreamsGroupTopologyDescriptionResponseStreamsGroupTopologyDescriptionUpdateResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The top-level error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+",
      "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." }
  ]
}

...

• 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(including an empty MemberId)
• UNSUPPORTED_VERSION — the coordinator cannot serve this RPC because no topology description plugin is configured
• STREAMS_TOPOLOGY_DESCRIPTION_TOO_LARGE — the plugin rejected the description because it exceeds the size the plugin is willing to storeTOPOLOGY_DESCRIPTION_UPDATE_UPDATE_FAILED — the plugin failed to process the request for some other reason; the client logs the underlying error at INFO level. The accompanying ErrorMessage carries the plugin's exception message. The broker's response shape is identical for both permanent and transient plugin failures; the distinction is broker-internal state that determines whether subsequent heartbeats at the same topology epoch will re-solicit (see Broker Side).

• UNKNOWN_MEMBER_ID — the member named in MemberId is no longer in the group; the client should treat itself as fenced and rejoin

• GROUP_ID_NOT_FOUND — the specified group 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

...

Client-side retry behavior for each code is described in Error Handling and Retries Client Side below.

StreamsGroupDescribeRequest Change

...

 {"name": "IncludeTopologyDescription", "type": "bool", "versions": "N+", "ignorable": true, "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 negotiatedClients on older versions never see the flag and behave unchanged.

StreamsGroupDescribeResponse Change

...

 { "name": "TopologyDescription", "type": "TopologyDescription", "versions": "N+",
  "nullableVersions": "N+", "ignorable": true, "default": "null",
  "about": "The topology description for this group. Null if not available — see TopologyDescriptionStatus for the reason." },
{ "name": "TopologyDescriptionStatus", "type": "int8", "versions": "N+", "ignorable": true, "default": "0",
  "about": "The 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), 3=AVAILABLE (a topology description is present in the TopologyDescription field)." }The broker MUST set this field to AVAILABLE whenever it attaches a TopologyDescription." }

The TopologyDescription common struct mirrors the struct used by StreamsGroupTopologyDescriptionUpdateRequest The TopologyDescription common struct mirrors the struct used by UpdateStreamsGroupTopologyDescriptionRequest (same field names and shape). Because  The nested struct names are prefixed TopologyDescription to avoid collision with the existing Subtopology struct already defined for the describe response. Because Kafka RPC schemas do not share common structs across message files, the struct is duplicated in StreamsGroupDescribeResponse.json.Setting  Setting these fields does not change the ErrorCode on the DescribedGroup: a group with a successful describe but a missing or failed topology fetch still returns ErrorCode=NONE. The TopologyDescriptionStatus field tells the caller why TopologyDescription is null, so that "waiting for first push" (NOT_STORED) can be distinguished from "broker-side fetch failed" (ERROR) without an error-level change to the describe result.

Plugin Interface

DeleteGroupsResponse Change

DeleteGroupsRequest and DeleteGroupsResponse are both bumped to the next version (3). The request shape is unchanged at the new version; the response adds an ErrorMessage field to each per-group DeletableGroupResultA new interface is introduced in org.apache.kafka.coordinator.group.api.streams :

{ "name": "ErrorMessage", "type": "string", "versions": "3+", "nullableVersions": "3+", "ignorable": true, "default": "null",  
  "about": "The error message, or null if there was no error." }  

A new generic error code is added to the per-group ErrorCode slot:

• GROUP_DELETION_FAILED — the delete operation could not complete; the accompanying ErrorMessage describes the underlying cause. The group is not tombstoned, and the caller may retry once the underlying condition is resolved. For streams groups configured with a topology description plugin this is returned when plugin.deleteTopology fails; other group types may adopt the same code in the future.

This is the only deletion-blocking failure mode introduced by this KIP. Consumer and share groups are unaffected. See Broker Side for the full ordering rule and the recovery path.

Plugin Interface

A new interface is introduced in org.apache.kafka.coordinator.group.api.streams :

/**
 * A broker-side plugin that stores, forwards, or exposes topology descriptions pushed
 * by Kafka Streams clients.
 *
 * 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.
 *
 * <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
 * {@code TopologyDescriptionRequired=true} in of the same heartbeat cycle; concurrent
 *group; calls with the same
 * {@code (groupId, topologyEpoch)} pair carry identical data
 * and must be treated as idempotent.
  */
public interface{@link StreamsGroupTopologyDescriptionPlugin#deleteTopology} extendsmust Configurable,also AutoCloseablebe {

idempotent — it may /**
be called more than once
 * Returns whetherfor the brokersame 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
     * sets {@code TopologyDescriptionRequired=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 groupCreationTimeMs the timestamp when the group was created. A value
     *                            of {@code 0} means "unset". Plugins should treat {@code 0}
     *                            as "incarnation indistinguishable" and should not cross-check
     *                            stored epoch values against another stored description that
     *                            also had {@code groupCreationTimeMs == 0}.
     * @param topologyEpoch the topology epoch
     * @return true if the broker should request a topology push from the client{@code groupId}, including when nothing is stored.
 */
@InterfaceStability.Evolving  
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 drives broker-side behaviour:
	 *
	 * <ul>
	 *   <li>{@link StreamsTopologyDescriptionPermanentFailureException} — the description will never be accepted
	 *       at this topology epoch (e.g. too large, semantically rejected). The broker
	 *       ratchets {@code LastFailedTopologyEpoch} and stops re-soliciting until the
	 *       epoch advances.</li>
	 *   <li>{@link StreamsTopologyDescriptionTransientFailureException} or any other exception — treated as
	 *       transient. The broker arms or extends the per-group back-off (30 s → 1 h,
	 *       exponential) and re-solicits on a later heartbeat.</li>
	 * </ul>
	 *
	 * In both cases the caller receives error code
	 * {@code STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED} with the exception's message in
	 * {@code ErrorMessage}; the permanent-vs-transient split is broker-internal state.
	 */
	CompletableFuture<Void> setTopology(String groupId, int topologyEpoch,
                                    	StreamsGroupTopologyDescription description);

	/**
	 * Remove any topology description stored for this group. Called when the group is
	 * deleted or expires. A failure (future completed exceptionally) is reported to the
	 * caller of {@code DeleteGroups} as {@code GROUP_DELETION_FAILED} with the exception message
	 * in the per-group {@code ErrorMessage}, and the broker does not tombstone the group;
 	 * a retry of {@code DeleteGroups} re-invokes this method idempotently. The
	 * periodic-cleanup path treats a failure identically — the group's tombstone is
	 * deferred to a future cycle.
	 */
	CompletableFuture<Void> deleteTopology(String groupId);

    /**
     * Return the stored topology description for {@code (groupId, topologyEpoch)}, or
     * {@code null} if the plugin no longer has the data (e.g. backend wipe). If the future
     * completes exceptionally, the broker reports a read error for the group.
     */
    booleanCompletableFuture<StreamsGroupTopologyDescription> requiresTopologyPushgetTopology(AuthorizableRequestContextString requestContextgroupId,
       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 final class Subtopology {
        public String groupId, long groupCreationTimeMs, int topologyEpoch);


id();
        public Collection<Node> nodes();
    }

	/**
     * A processing Callednode whenin athe clienttopology. sendsPredecessor anodes topologycan descriptionbe forinferred afrom streamssuccessor grouprelation.
     */
  This  methodpublic maysealed beinterface calledNode concurrently{
 by multiple members of the same group;
 String  name();
  *  all calls for the sameSet<String> successors(groupId, topologyEpoch) carry identical data.);
    }

    public *
static final class Source implements *Node <p>The{
 returned future completes when the topology has beenpublic persisted orSet<String> topics();
    }

 * forwarded. All failurespublic muststatic befinal signalledclass byProcessor completingimplements theNode returned future{
     * exceptionally — implementationspublic must not throw synchronously from this method.Set<String> stores();
    }

    public *static Thefinal brokerclass handlesSink theimplements future's completion exception as follows:
     * {@link org.apache.kafka.common.errors.InvalidRequestException} maps to
     * {@code INVALID_REQUEST}; {@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeException}
Node {
        public Optional<String> topic();
    }

    public static final class GlobalStore {
      * maps topublic {@code TOPOLOGY_DESCRIPTION_TOO_LARGE}; any other exception maps to
Source source();
        public  * {@code TOPOLOGY_DESCRIPTION_UPDATE_FAILED} and is logged at WARN.Processor processor();
    }
}

Two new exception classes in the same package let the plugin signal the permanent-vs-transient distinction. Plugins that throw any other exception are treated as transient:

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

import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.annotation.InterfaceStability;

/** Signals that the topology description for the current epoch will never be accepted (e.g. too large, semantically rejected). */
@InterfaceStability.Evolving  
public class StreamsTopologyDescriptionPermanentFailureException extends ApiException {
    public StreamsTopologyDescriptionPermanentFailureException(String message) { super(message); }
    public StreamsTopologyDescriptionPermanentFailureException(String message, Throwable cause) { super(message, cause); }
}

/** Signals a transient backend failure; the broker re-solicits on a later heartbeat. Plugins that throw any other exception are treated identically. */
@InterfaceStability.Evolving  
public class StreamsTopologyDescriptionTransientFailureException extends ApiException {
    public StreamsTopologyDescriptionTransientFailureException(String message) { super(message); }
    public StreamsTopologyDescriptionTransientFailureException(String message, Throwable cause) { super(message, cause); }
}

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 (signalled by StreamsTopologyDescriptionPermanentFailureException), 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 new enum { NOT_REQUESTED, NOT_STORED, ERROR, AVAILABLE }. Each enum value's ordinal matches the wire-level TopologyDescriptionStatus int8 (0, 1, 2, 3). AVAILABLE is reported when TopologyDescription is non-null.

The Admin client exposes a POJO hierarchy in org.apache.kafka.clients.admin that mirrors org.apache.kafka.streams.TopologyDescription but lives in the clients module. The admin client converts the wire-format struct into this hierarchy.

package org.apache.kafka.clients.admin;

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

    public static final 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 final class Source implements Node {
        public Set<String> topics();
    }

    public static final class Processor implements Node { *
     * @param requestContext the context of the UpdateStreamsGroupTopologyDescription request
     * @param groupId the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created
     * @param topologyEpoch the topology epoch
     * @param description the topology description
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> setTopology(AuthorizableRequestContext requestContext,
                                        String groupId, long groupCreationTimeMs, int topologyEpoch,
                                        StreamsGroupTopologyDescription description);

    /**
     * Called when a group is explicitly deleted via DeleteGroups. Removes any topology
     * description stored for this group.
     *
     * <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 DeleteGroups response returned to the caller.
     *
     * @param requestContext the context of the DeleteGroups request
     * @param groupId the streams group ID
     * @return a future that completes when the operation is done
     */
    CompletableFuture<Void> deleteTopology(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
     * {@code IncludeTopologyDescription=true}.
     *
     * <p>Returns a future that resolves to the stored topology description for the
     * given {@code (groupId, groupCreationTimeMs, topologyEpoch)} tuple, or to
     * {@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 completes
     * exceptionally, the plugin signals a read error for this group.
     *
     * @param requestContext the context of the StreamsGroupDescribe request
     * @param groupId the streams group ID
     * @param groupCreationTimeMs the timestamp when the group was created
     * @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(AuthorizableRequestContext requestContext,
        public Set<String> stores();
    }

    public static Stringfinal groupId,class longSink groupCreationTimeMs,implements int topologyEpochNode {
        public Optional<String> topic();
    }

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.

...

...

    public static final 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 StreamsGroupTopologyDescriptionUpdate 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 StreamsGroupTopologyDescriptionUpdate, 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 StreamsTopologyDescriptionPermanentFailureException it writes LastFailedTopologyEpoch = pushedEpoch so subsequent heartbeats at the same epoch do not re-solicit. On StreamsTopologyDescriptionTransientFailureException or any other exception it writes no metadata record, arms the per-group back-off, and the next heartbeat re-solicits once the window elapses. In both failure cases the response carries STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED with the plugin's exception message in ErrorMessage; the permanent-vs-transient split is broker-internal state. 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. On plugin success the group is tombstoned and the per-group ErrorCode is NONE. On plugin failure the group is not tombstoned and the per-group ErrorCode is set to GROUP_DELETION_FAILED with the plugin's exception message in ErrorMessage (also logged at WARN); the operator retries the request once the plugin recovers, or unsets the plugin config to bypass it. Other groups in the same batch are unaffected — the failure is reported per group. This ordering matches the natural-expiration cleanup below, which also defers tombstoning until plugin.deleteTopology succeeds.

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 StreamsGroupTopologyDescriptionUpdate 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 member has been dropped from the group: 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 (STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the flag and log at WARN with the response's ErrorMessage; 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 StreamsTopologyDescriptionPermanentFailureException; the broker persists the rejection at the epoch level via LastFailedTopologyEpoch and stops re-soliciting at the same epoch. The exception message reaches the client in ErrorMessage.
• Signal transient storage-layer failures by completing the setTopology future with StreamsTopologyDescriptionTransientFailureException (or 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 StreamsGroupTopologyDescriptionUpdate 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.

DeleteGroupsRequest and DeleteGroupsResponse are bumped to version 3, which adds a per-group ErrorMessage field on the response and introduces the new GROUP_DELETION_FAILED error code. Older admin clients negotiate version 2 and never see the new field; they continue to receive only the per-group ErrorCode. DeleteGroups can newly fail on streams groups when a topology description plugin is configured and its deleteTopology call fails. Brokers without a configured plugin keep today's behaviour. Older clients that do receive GROUP_DELETION_FAILED (because they understand version 3 but predate this KIP's error-code addition) decode the code as UNKNOWN_SERVER_ERROR via the standard forward-compatibility fallback in Errors.forCode; the group is still not tombstoned, so an idempotent retry of DeleteGroups converges once the plugin recovers. No client-side change is required

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

The Admin client exposes a POJO hierarchy in org.apache.kafka.clients.admin that mirrors org.apache.kafka.streams.TopologyDescription but lives in the clients module. The admin client converts the wire-format struct into this hierarchy.

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

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.

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

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

...

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

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

System Tests

Test Plan

Integration Tests

Broker

• StreamsGroupTopologyDescriptionUpdateRequestTest (new) — push happy path; StreamsTopologyDescriptionPermanentFailureException (ratchets LastFailedTopologyEpoch) and StreamsTopologyDescriptionTransientFailureException (arms back-off) both surface as STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED with the cause in ErrorMessage; heartbeat-flag gating; member/group/MemberId fencing; explicit DeleteGroups plugin-success path (tombstone written) and plugin-failure path (GROUP_DELETION_FAILED returned with ErrorMessage, group not tombstoned, retry converges after plugin recovery).
• StreamsGroupTopologyDescriptionUpdateNoPluginRequestTest (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).

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

...

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [orders])
      --> KSTREAM-FILTER-0000000001
    Processor: KSTREAM-FILTER-0000000001 (stores: [])
      --> KSTREAM-SINK-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Sink: KSTREAM-SINK-0000000002 (topic: valid-orders)
      <-- KSTREAM-FILTER-0000000001

The corresponding UpdateStreamsGroupTopologyDescription StreamsGroupTopologyDescriptionUpdate request body, with the topology converted to the wire format, looks like this:

...