...

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

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

...

{
  "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 (including an empty MemberId), or the plugin semantically rejected the payload by completing its future with InvalidRequestException

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

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

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

...

The TopologyDescription common struct mirrors the struct used by UpdateStreamsGroupTopologyDescriptionRequest StreamsGroupTopologyDescriptionUpdateRequest (same field names and shape). Because Kafka RPC schemas do not share common structs across message files, the struct is duplicated in StreamsGroupDescribeResponse.json. 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

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

DeleteGroupsResponse Change

No schema change. A new error code is added to the per-group ErrorCode slot of the existing DeleteGroupsResponse:

• STREAMS_TOPOLOGY_DESCRIPTION_DELETE_FAILED — the topology description plugin failed to delete the description for this streams group; the group is not tombstoned. The caller may retry the request once the plugin recovers, or unset group.streams.topology.description.plugin.class to bypass the plugin. The plugin's exception is logged at WARN on the broker; the response itself carries only the error code, matching the existing DeleteGroupsResponse shape.

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.
 *
 * <p>Implementations must be thread-safe. {@link #setTopology} may be called
 * concurrently by multiple members of the same group; calls with the same
 * {@code (groupId, topologyEpoch)} carry identical data and must be idempotent.
 * {@link #deleteTopology} must also be idempotent — it may be called more than once
 * for the same {@code groupId}, including when nothing is stored.
 */
public interface StreamsGroupTopologyDescriptionPlugin extends Configurable, AutoCloseable {

    /**
     * Store the topology description for a streams group.
     *
     * <p>The returned future completes when the topology has been persisted or forwarded.
     * Failures must be signalled by completing the future exceptionally — implementations
     * must not throw synchronously. The completion exception maps to the client-visible
     * error code:
     *
     * <ul>
     *   <li>{@link org.apache.kafka.common.errors.InvalidRequestException} — payloads the
     *       plugin will not accept on semantic grounds; reported as {@code INVALID_REQUEST}.</li>
     *   <li>{@link org.apache.kafka.common.errors.TopologyDescriptionTooLargeExceptionStreamsTopologyDescriptionTooLargeException} —
     *       descriptions larger than the plugin is willing to store; reported as
     *       {@code STREAMS_TOPOLOGY_DESCRIPTION_TOO_LARGE}.</li>
     *   <li>Any other exception — transient backend failure; reported as
     *       {@code STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED}.</li>
     * </ul>
     *
     * The first two are treated as permanent at this topology epoch and no further push
     * will be solicited until the epoch advances. The third is treated as transient and
     * may be retried.
     */
    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. FailuresA arefailure logged(future bycompleted theexceptionally) broker but do not propagateis reported to the
	 * caller of {@code DeleteGroups} as {@code STREAMS_TOPOLOGY_DESCRIPTION_DELETE_FAILED} on that
	 * user-visiblegroup's per-group result ofand whateverthe operationbroker triggereddoes thenot removal.
tombstone the group; a retry */of
	 * {@code DeleteGroups} CompletableFuture<Void> deleteTopology(String groupId);

    /**
     * Return the stored 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.
     */
    CompletableFuture<StreamsGroupTopologyDescription> getTopology(String groupId, int topologyEpoch);
}

...

1. The plugin is instantiated at broker startup if group.streams.topology.description.plugin.class is configured. A broker without a plugin returns UNSUPPORTED_VERSION for UpdateStreamsGroupTopologyDescription 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 UpdateStreamsGroupTopologyDescription 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 InvalidRequestException or TopologyDescriptionTooLargeException StreamsTopologyDescriptionTooLargeException it writes LastFailedTopologyEpoch = pushedEpoch so subsequent heartbeats at the same epoch do not re-solicit. Any other exception maps to STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED, is logged at WARN, and is treated as transient — no metadata record is written and the next heartbeat re-solicits. If the plugin call succeeds but the metadata-record write fails, the next heartbeat sees the drift, re-solicits, and the idempotent re-push closes the gap.

4. On On DeleteGroups, the broker calls deleteTopology on the plugin before writing the group tombstone, for each requested streams group with StoredTopologyEpoch != -1. Plugin failures are logged but do not affect the deletion response; the group is tombstoned regardless. 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 STREAMS_TOPOLOGY_DESCRIPTION_DELETE_FAILED (the plugin's exception is logged at WARN on the broker); 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 belowcleanup 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.

...

1. At startup, if topology.description.push.enabled=true, the Streams client converts the topology returned by Topology#describe() to the wire format and stores it internally. The mapping is one-to-one with the RPC schema; predecessor edges are not sent on the wire and the read side reconstructs them by inverting each node's successor list. When topology.description.push.enabled=false, no description is stored and the feature is disabled on this client.
2. The Streams client records the TopologyDescriptionRequired flag from each heartbeat response.
3. On each consumer background-thread poll, the client sends UpdateStreamsGroupTopologyDescription 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 broker no longer recognizes this member (group deleted, or member dropped): the client clears the flag and relies on the existing membership-management path to trigger a clean rejoin on the next heartbeat. All other errors (STREAMS_TOPOLOGY_DESCRIPTION_TOO_LARGE, STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED, INVALID_REQUEST, UNSUPPORTED_VERSION, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the flag and log at WARN; the client does not retry on its own, and a re-attempt happens only when the broker re-sets the flag via a subsequent heartbeat. A non-zero ThrottleTimeMs on the response delays the next push attempt by that amount, as with other request managers.

...

• Treat setTopology (on (groupId, topologyEpoch)) and deleteTopology (on (groupId)) as idempotent; the broker may re-issue an identical call when an earlier call's bookkeeping write failed.
• Be thread-safe under concurrent invocation: setTopology may be called by multiple members of the same group in the same heartbeat cycle, and the periodic-cleanup path may invoke deleteTopology while a member is mid-push.
• Reject payloads the plugin will not accept by completing the setTopology future with TopologyDescriptionTooLargeException StreamsTopologyDescriptionTooLargeException or InvalidRequestException; the broker persists the rejection at the epoch level via LastFailedTopologyEpoch and stops re-soliciting at the same epoch.
• Signal transient storage-layer failures by completing the setTopology future with any other exception; the broker's per-group back-off (30 s → 1 h, exponential) throttles re-solicitation.

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

• Avoid blocking coordinator threads (plugin methods may be invoked on them) and complete futures within seconds, not minutes; the broker applies no wall-clock deadline to plugin calls, so the coordinator's responsiveness is bounded by what the plugin does. Bound plugin-side state explicitly — the plugin shares the broker heap.

...

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 StreamsGroupTopologyDescriptionUpdate path is guarded by READ on the GROUP, identical to the existing heartbeat ACL.

...

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.

...

Test Plan

Integration Tests

Broker:

• UpdateStreamsGroupTopologyDescriptionRequestTest

  StreamsGroupTopologyDescriptionUpdateRequestTest (new) — push happy path; permanent and transient plugin failures; heartbeat-flag gating; member/group/MemberId fencing; explicit DeleteGroups

  ordering.

  plugin-success path (tombstone written) and plugin-failure path (STREAMS_TOPOLOGY_DESCRIPTION_DELETE_FAILED returned, group not tombstoned, retry converges after plugin recovery).

• StreamsGroupTopologyDescriptionUpdateNoPluginRequestTest UpdateStreamsGroupTopologyDescriptionNoPluginRequestTest (new) — broker without plugin: pushes rejected, flag never set, describe returns NOT_STORED.
• AuthorizerIntegrationTest — READ ACL on the GROUP resource for the new RPC.

...

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:

...