Page History

...

Group ID, member ID, member epoch, and instance ID (if defined by the client) are sent with each heartbeat request. Any other information that has not changed since the last heartbeat can be omitted, unless an error occurred.

Heartbeat handling & response

...

It is up to the client implementation how topology IDs are defined. The protocol and the broker only requires that two clients with the same topology ID should have the same topology. In the initial implementation, the topology ID will be derived from the topology by deterministic hashing. However, it would also be possible to derive an implementation where topology IDs are explicitely defined through a configuration. This would, for example, allow the topologies to be manually versioned or topology IDs be derived from a git commit hash.

We also allow the StreamsGroupInitialize to be invoked by a client that is not a member of the group and was not asked to initialize the group. This can also be used to create an empty initialized group. However, default mode is that a group is initialized by one of the clients after joining. Adding a strict mode, where topology updates are requested explicitly, any implicit updates of the topology are topology are rejected and members with incompatible topologies members with incompatible topologies are fenced out of the group can be built on top of this protocol, but will be postponed to a follow-up KIP.

...

The possible states of the streams group are EMPTY, ASSIGNING, RECONCILING , STABLE, DEAD as for consumer groups, and additionally INITIALIZING for streams groups that have members but are pending initialization with a topology through the StreamsGroupInitialize RPC.

...

The StreamsGroupInitialize API is called to initialize the topology information on the broker side, that will be reused by all members of the group. The StreamsGroupInitialize RPC can be is normally called only by a single application to explicitly initialize the group, after the broker requested the initialization from that member. It can, however, also be used to create and initialize an empty group.

Request Schema

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "StreamsGroupInitializeRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
      "about": "The group identifier." },
    { "name":  "Topology", "type": "[]Subtopology", "versions": "0+",
      "about": "The sub-topologies of the streams application.",
      "fields": [
        { "name": "Subtopology", "type": "string", "versions": "0+",
          "about": "String to uniquely identify the sub-topology. Deterministically generated from the topology" },
        { "name": "SourceTopics", "type": "[]string", "versions": "0+",
          "about": "The topics the topology reads from." },
        { "name": "SourceTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
           "about": "The regular expressions identifying topics the sub-topology reads from. null if not provided." },
        { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of state changelog topics associated with this sub-topology. Created automatically." },
        { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
          "about": "The repartition topics the sub-topology writes to." }, 
        { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
          "about": "The set of source topics that are internally created repartition topics. Created automatically." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "TopicConfig", "versions": "0+", "fields": [
        { "name": "key", "type": "string", "versions": "0+",
          "about": "The key of the topic-level configuration." },
        { "name": "value", "type": "string", "versions": "0+",
          "about": "The value of the topic-level configuration," }
      ]
    },
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The name of the topic." },
      { "name": "Partitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions in the topic. Can be 0 if no specific number of partitions is enforced. Always 0 for changelog topics." },
      { "name": "TopicConfigs", "type": "[]TopicConfig", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

When the group coordinator handles a StreamsGroupInitialize request:

• Looks up the group.

• It performs request validation.
• Checks if the provided group existsLooks up or creates the group. GROUP_ID_NOT_FOUND is returned if the group does not exist or if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).

• Creates all required internal topics.

  • STREAMS_MISSING_SOURCE_TOPICS is returned if there are source topics missing for a topology that is supposed to be initialized. Also returned if the source topic regular expression matched no topics.

  • STREAMS_INCONSISTENT_INTERNAL_TOPICS is returned if there are internal topics present on the broker that are not consistent with the internal topic requirements of the provided topology.

• Writes the topology, keyed with the groupId to the consumer offset topic. Existing records will be overwritten.

...

• When the first member joins the group, the group metadata and the member metadata records for the member are first created.
• Every time a new member joins the group, another member metadata record is added
• When a previously non-existing group is initialized via a StreamsGroupInitialize  RPC, the group member metadata is created.
• Whenever a member updates its member metadata through the heartbeat, a new version of the member metadata record is appended.
• When a member leaves the group, a tombstone is written for the member metadata key.

When the group lost all members (or never had any)  and offsets.retention.minutes expires, the group metadata tombstone is written. This can happen in both states EMPTY and INITIALIZING.

StreamsGroupMemberMetadataKey

...

When the group lost all members (or never had any) and offsets.retention.minutes expires, tombstones is written to remove the records.

...

The target assignment is stored in N + 1 records, where N is the number of members in the group. When a new target assignment is computed, the group coordinator will compare it with the current target assignment and only write the difference between the two assignments to the __consumer_offsets  partition. That also means if a member left the group, a corresponding tombstone record is written. The assignment must be atomic, so the group coordinator will ensure that all the records are written in a single batch, limiting the size to the maximum batch size (default 1MB), as in consumer groups. When the group lost all members and offsets.retention.minutes expires, the group metadata tombstone is written. 

StreamsGroupTargetAssignmentMemberKey

...