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

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

 { "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();
    }
}