Page History

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

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

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

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

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

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 id();
        public Collection<Node> nodes();
    }

	/**
     * A processing node in the topology. Predecessor nodes can be inferred from successor relation.
     */
    public sealed interface Node {
        String name();
        Set<String> successors();
    }

    public static final class Source implements Node {
        public Set<String> topics();
    }

    public static final class Processor implements Node {
        public Set<String> stores();
    }

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

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

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

 { "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 (TooLargesignalled /by InvalidRequestStreamsTopologyDescriptionPermanentFailureException), or -1 if none. Heartbeat-path solicitation is suppressed while this equals the current topology epoch, to avoid hot-looping." }

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 {
        public Set<String> stores();
    }

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

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