...

This KIP proposes a mechanism to send the full topology description from the client to the broker, where a pluggable backend can store and expose it for operational tooling and topology visualization in management UIs. The design follows the pattern established by KIP-714 (Client Metrics and Observability): the broker acts as a conduit, receiving data from clients and delegating storage and presentation to a plugin implementation. This keeps the broker itself simple and the feature extensible.

Public Interfaces

New Broker Configuration

New Client Configuration

StreamsGroupHeartbeatResponse Change

StreamsGroupHeartbeatResponse is bumped to the next version (N) and gains a new field at that version:

 { "name": "TopologyDescriptionRequiredTopologyDescriptionId", "type": "booluuid", "versions": "N+",
  "nullableVersions": "N+", "default": "falsenull",
  "about": "True ifWhen non-null, the broker'sclient topologyshould descriptionpush plugin does not have an up-to-dateits current topology description tagged forwith this group.id Thevia clientUpdateStreamsGroupTopologyDescription. shouldNull sendwhen theno topologypush description via UpdateStreamsGroupTopologyDescription."is requested." }

The broker sets this field to true when a topology description plugin is configured and 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 plugin.requiresTopologyPush(requestContext, groupId, groupCreationTimeMs, topologyEpoch)topologyDescriptionId) with the current id; if the plugin returns true, where requestContext is the context of the heartbeat requestthe broker includes that id in the heartbeat response, otherwise the field is omitted.

New RPC: UpdateStreamsGroupTopologyDescription (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.

Request:

{
  "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": "TopologyEpochTopologyDescriptionId", "type": "int32uuid", "versions": "0+",
      "about": "The epochtopology ofdescription the topology being describedid received in the most recent heartbeat response, identifying which topology version this push corresponds to." },
    { "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+",
        "about": "The source topics this node reads from. Populated for source nodes." },
      { "name": "SinkTopic", "type": "string", "versions": "0+",
        "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." }
    ]}
  ]
}

Response:

{
  "apiKey": "TBD",
  "type": "response",
  "name": "UpdateStreamsGroupTopologyDescriptionResponse",
  "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." }
  ]
}

...

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

StreamsGroupDescribeRequest Change

StreamsGroupDescribeRequest is bumped to the next version (N) and gains a new field at that version:

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

A client that negotiates an older version is handled unchanged by any broker. The flag may only be set when version N or later is negotiated.

StreamsGroupDescribeResponse Change

StreamsGroupDescribeResponse is bumped to the next version (N). Two new fields are added to each DescribedGroup at that version:

 { "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 reason: 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." }
 

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.

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.

...

package org.apache.kafka.server.streams;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.UpdateStreamsGroupTopologyDescriptionRequestData;
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 TopologyDescriptionRequired=trueTopologyDescriptionId} in the sametheir heartbeat cycleresponses; concurrent
 * calls with the same {@code (groupId, topologyEpochtopologyDescriptionId)} 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 setsthe {@code TopologyDescriptionRequired=truetopologyDescriptionId} 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 groupCreationTimeMstopologyDescriptionId the broker-minted timestampid whenidentifying the group was created. A valuecurrent topology
     *        ofversion {@codefor 0}this means "unset". Plugins should treat {@code 0}group; opaque to the plugin
     * @return true if the broker should request asa "incarnationtopology indistinguishable"push andfrom shouldthe not cross-check stored
 client
     */
    boolean requiresTopologyPush(AuthorizableRequestContext requestContext,
  epoch values against another stored description that also had
     *        {@code groupCreationTimeMs == 0}.
     * @param topologyEpochString thegroupId, topology epochUuid topologyDescriptionId);

     /**
 @return true if the broker* shouldCalled requestwhen a topologyclient pushsends froma thetopology client
description for a streams  */group.
    boolean requiresTopologyPush(AuthorizableRequestContext requestContext,
            * This method may be called concurrently by multiple members of the same group;
     * all calls for the            String same (groupId, longtopologyDescriptionId) groupCreationTimeMs,carry int topologyEpoch);

identical data.
     /**
     * Called<p>The whenreturned afuture clientcompletes sendswhen athe topology descriptionhas forbeen a streams group.persisted or
     * Thisforwarded. methodComplete mayit be called concurrently by multiple members of the same group;exceptionally with
     * {@link org.apache.kafka.common.errors.InvalidRequestException} to signal that
     * the payload allis callssemantically forinvalid; theany sameother (groupId, topologyEpoch) carry identical data.exception is treated as a
     * server-side plugin error.
     *
   <p>The returned future* completes@param whenrequestContext the topologycontext hasof beenthe persistedUpdateStreamsGroupTopologyDescription orrequest
     * @param forwarded.groupId Completethe itstreams exceptionallygroup withID
     * @param {@link org.apache.kafka.common.errors.InvalidRequestException} to signal that
     * the payload is semantically invalid; any other exception is treated as a
     * server-side plugin error.
     *topologyDescriptionId the id this push is tagged with, as carried in the
     *        heartbeat response that asked for it; opaque to the plugin
     * @param requestContextdescription the topology contextdescription
 of the UpdateStreamsGroupTopologyDescription request
 * @return a future *that @paramcompletes groupIdwhen the streamsoperation groupis IDdone
     */
 @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 topologyEpochUuid topologyDescriptionId,
                                        UpdateStreamsGroupTopologyDescriptionRequestData.TopologyDescription description);

    /**
     * Called when a group is explicitly deleted via DeleteGroups. Removes the topology
     * for all epochs.
     *
     * <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, topologyEpochtopologyDescriptionId)} tuplepair, or to
     * {@code null} if no
     * topology is stored (e.g. no push has succeeded yet, or the stored description
     * storedis descriptiontagged is forwith a different epochid). If the future completes exceptionally, the
     * the plugin signals a read error for this group.
     *
     * @param requestContext the context of the StreamsGroupDescribe request
     * @param groupId the streams group ID
     * @param groupCreationTimeMstopologyDescriptionId the timestampid whenof the topology version groupthe wascaller createdis
     * @param topologyEpoch the topology epoch the caller is asking about; opaque to the plugin
     * @return a future resolving to the stored topology description, or null if none
     */
    CompletableFuture<UpdateStreamsGroupTopologyDescriptionRequestData.TopologyDescription>
        getTopology(AuthorizableRequestContext requestContext,
                    String groupId, long groupCreationTimeMs, int topologyEpochUuid topologyDescriptionId);
}

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.

...

The Admin client exposes a POJO hierarchy in org.apache.kafka.clients.admin rather than the generated wire-format class:

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();
        Set<String> predecessors();
        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();
    }
}

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

...

1. The plugin is instantiated at broker startup if group.streams.topology.description.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 includes a TopologyDescriptionId in heartbeat responses, in normal operation the RPC is only sent against a broker that has a plugin configured.
2. 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. 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 sets TopologyDescriptionRequired=true on includes the id in 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 groupCreationTimeMs parameter identifies the specific incarnation of a group; a recreated group with the same groupId gets a fresh creation timestampplugin sees only the opaque id and the group identity; it does not see topology epochs or wall-clock timestamps.
3. On UpdateStreamsGroupTopologyDescription, the broker . 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.On UpdateStreamsGroupTopologyDescription, the broker checks the READ ACL on the group and that a plugin is configured. If the serialized description exceeds group.streams.topology.description.max.bytes, the request is rejected with TOPOLOGY_DESCRIPTION_TOO_LARGE. Size is measured at the highest UpdateStreamsGroupTopologyDescription version the broker supports — not the negotiated version — so a pre-validating client should use the same version (discoverable via ApiVersions). The broker then 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, any other exception maps to UNKNOWN_SERVER_ERROR, and both 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. 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).

...

1. The Streams client records the TopologyDescriptionRequired flag TopologyDescriptionId 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 when (carrying the most recently received TopologyDescriptionId) when a coordinator is known, the flag id is setnon-null, a stored topology description is available, and no prior request is in flight. Completion handling is described in Error Handling and Retries.

...

NOT_COORDINATOR and COORDINATOR_NOT_AVAILABLE trigger coordinator rediscovery; the flag stays setpending id is preserved. COORDINATOR_LOAD_IN_PROGRESS and network exceptions leave the flag set pending id in place for retry on the next poll.

All other errors (TOPOLOGY_DESCRIPTION_TOO_LARGE, INVALID_REQUEST, UNSUPPORTED_VERSION, UNKNOWN_SERVER_ERROR, GROUP_ID_NOT_FOUND, GROUP_AUTHORIZATION_FAILED) clear the topologyDescriptionRequired flag pending id 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 sends a non-null id on 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 outdated epochs superseded ids internally. Incoming requiresTopologyPush calls can be used as an implicit keep-alive.
• Avoid concurrent or repetitive pushes: track in-flight pushes per groupid, back off with an exponential schedule on transient failures, and disable pushes permanently for the group id on permanent failures (TOPOLOGY_DESCRIPTION_TOO_LARGE, plugin-semantic INVALID_REQUEST).
• Not request repeated pushes from stable groups that have already pushed their topology at under the current epochid.

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 bumps StreamsGroupHeartbeatResponse, StreamsGroupDescribeRequest, and StreamsGroupDescribeResponse to the next available version of each RPC and adds the new fields (TopologyDescriptionRequiredTopologyDescriptionId, 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 TopologyDescriptionRequired flag TopologyDescriptionId is only set by plugin-equipped coordinators; a client whose current coordinator lacks the plugin never sees the flagid. If the coordinator migrates mid-push to a plugin-less broker, the client receives UNSUPPORTED_VERSION, clears the topologyDescriptionRequired flagpending id, and does not retry. The flag is set A new id is sent again only if when a future heartbeat response from a plugin-equipped coordinator includes TopologyDescriptionRequired=trueone.

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 -epoch topology remains uncaptured until at least one member heartbeats with at the new epoch.

Future Work

Hash-based mismatch detection. A future enhancement could introduce a topology hash to detect clients on running different topology descriptions reporting code at the same topology epoch (e.g., a partial deployment), which the broker-minted id alone cannot catch.

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.

...

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 are expected to be well fit comfortably under the 1 MB 350 KB default limit even for complex 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:

StreamsBuilder builder = new StreamsBuilder();
builder.stream("orders")
       .filter((key, value) -> value != null)
       .to("valid-orders");

Topology#describe() produces the following text representation:

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 request body, with the topology converted to the wire format, looks like this:

{
  "GroupId": "orders-app",
  "TopologyDescriptionId": "8d7e3c2a-4f6b-4d9a-9c1b-7e5f3a8d2c4e",
  "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": []
  }
}

Predecessor edges (<-- in the text representation) are not sent on the wire; the read side reconstructs them by inverting each node's Successors.