Page History

...

All non-assignment metadata as defined in the ConsumerGroupHeartbeatRequest such as the group ID, member ID, member epoch, instance ID, rack ID and rebalance timeout.
The currently assigned active, standby and warm-up tasks. Each task is identified by a subtopology ID (which can be an arbitrary string in protocol for future extensibility, even though subtopologies are currently numbered) and a partition number.
The process ID to identify multiple members running in the same process (i.e. the same Streams client).
The user-defined endpoint to be used for running interactive queries.
Topology metadata needed for creating internal topics and computing the assignment.
The topology epoch, which is defined by the client, and will be 1 normally0 by default. The epoch can be bumped by the client to deploy a new version of the topology metadata.
For each task, the current sum of task changelog offsets and task changelog end offsets. These are used to determine whether a warm-up task is below acceptable.recovery.lag and can be used to optimize the assignment in the assignor. For more details on these offsets see "Cumulative task changelog offsets and task changelog end offsets" section below
A flag to request shutdown of the whole application (only if the member epoch is the group leave epoch).

...

The metadata sent to the broker via the heartbeat is persisted as a new record type in the __consumer_offsets topic and is part of the group coordinator state in subsequent requests. If one or more source topics or internal topics do not exist, the group will enter state NOT_READY and will attempt to create any missing internal topics (see more details below). Only after the topology metadata is initialized and all source topics and internal topics exist, the group coordinator starts assigning tasks to the members of the group.

Topology updates

To avoid unintential unintentional topology updates or rollbacks, updating the broker-side topology after it has been initialized for a group requires bumping the client-side configuration topology.epoch, which defaults to 10. By default, whenever a Streams member joins the group with a topology that is different from the current topology of the streams group, and the topology epoch is not bumped, the group coordinator will respond with an error, indicating that the member is incorrectly updating the broker-side topology.

...

For active processing tasks, the cumulative task changelog end offset is the same as the cumulative task changelog offset
For all restoring tasks (active, standby or warm-up), the cumulative end-offset is the sum of the last end offsets cached by the restore consumer. If an end offset of a topic partition is unknown, no end-offset is reported.
For dormant standby tasks, no end-offset is reported.

...

Performs request validation.
If the member joins the group (i.e. member epoch is 0):
1. Look up or create the group.
  1. If the group is created the topology epoch of the group is set to the topology epoch sent by the member.
2. GROUP_ID_NOT_FOUND is returned if the group ID is associated with a group type that is not streams or classic (the latter will be allowed for migration).
3. Compare the topology epoch of the request ER to the topology epoch of the group EG
  1. If ER=EG, check if topology metadata in the request is equal to the topology metadata of the group, otherwise fail with STREAMS_INVALID_TOPOLOGY_EPOCH
  2. If ER<EG, fail with STREAMS_TOPOLOGY_FENCED
  3. If ER>EG+1, fail with STREAMS_INVALID_TOPOLOGY_EPOCH
  4. If ER=EG+1, update the topology by writing the new topology record to the offset topic
4. Creates the member.
If the member is already part of the group (i.e. member epoch is greater than 0):
1. Looks up the group.
2. GROUP_ID_NOT_FOUND is returned if the group ID does not exist anymore.
3. If the member does not exist, returns UNKNOWN_MEMBER_ID
4. Checks whether the member epoch matches the member epoch in its current assignment. FENCED_MEMBER_EPOCH is returned otherwise. The member is also removed from the group.
  - There is an edge case here. When the group coordinator transitions a member to its target epoch, the heartbeat response with the new member epoch may be lost. In this case, the member will retry with the member epoch that it knows about and its request will be rejected with a FENCED_MEMBER_EPOCH. This will be handled as in KIP-848.
5. If the topology epoch in the request is less than the topology epoch of the group, set state STALE_TOPOLOGY in the response.
Updates information of the member if needed. The group epoch is incremented if there is any change.
If the topology or topic metadata changed, detect any topology / topic mismatches as described earlier in this document.
Reconcile the member assignments as explained earlier in this document.

...

Code Block

language	yml
linenumbers	true

{
  "apiKey": TBD,
  "type": "response",
  "name": "StreamsGroupDescribeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  // - NOT_COORDINATOR (version 0+)
  // - COORDINATOR_NOT_AVAILABLE (version 0+)
  // - COORDINATOR_LOAD_IN_PROGRESS (version 0+)
  // - INVALID_REQUEST (version 0+)
  // - INVALID_GROUP_ID (version 0+)
  // - GROUP_ID_NOT_FOUND (version 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": "Groups", "type": "[]DescribedGroup", "versions": "0+",
      "about": "Each described group.",
      "fields": [
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The describe error, 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." },
        { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
          "about": "The group ID string." },
        { "name": "GroupState", "type": "string", "versions": "0+",
          "about": "The group state string, or the empty string." },
        { "name": "GroupEpoch", "type": "int32", "versions": "0+",
          "about": "The group epoch." },
        { "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
          "about": "The assignment epoch." },

        { "name":  "Topology", "type": "Topology", "versions": "0+",
          "about": "The topology metadata currently initialized for the streams application.",
          "fields": [
            { "name": "Epoch", "type": "int32", "versions": "0+",
              "about": "The epoch of the currently initialized topology for this group." },
            { "name": "Subtopologies", "type": "[]Subtopology", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The subtopologies of the streams application. This contains the configured subtopologies, where the number of partitions are set and any regular expressions are resolved to actual topics. Null if the group is uninitialized, source topics are missing or incorrectly partitioned.",
              "fields": [
                { "name": "SubtopologyId", "type": "string", "versions": "0+",
                  "about": "String to uniquely identify the subtopology." },
                { "name": "SourceTopics", "type": "[]string", "versions": "0+",
                  "about": "The topics the subtopology reads from." },
                { "name": "RepartitionSinkTopics", "type": "[]string", "versions": "0+",
                  "about": "The repartition topics the subtopology writes to." },
                { "name": "StateChangelogTopics", "type": "[]TopicInfo", "versions": "0+",
                  "about": "The set of state changelog topics associated with this subtopology. Created automatically." },
                { "name": "RepartitionSourceTopics", "type": "[]TopicInfo", "versions": "0+",
                  "about": "The set of source topics that are internally created repartition topics. Created automatically." }
              ]}
          ]},
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The member epoch." },
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID for static membership." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The rack ID." },

            { "name": "ClientId", "type": "string", "versions": "0+",
              "about": "The client ID." },
            { "name": "ClientHost", "type": "string", "versions": "0+",
              "about": "The client host." },

            { "name": "TopologyEpoch", "type": "int32", "versions": "0+",
              "about": "The epoch of the topology on the client." },

            { "name": "ProcessId", "type": "string", "versions": "0+",
              "about": "Identity of the streams instance that may have multiple clients. " },
            { "name": "UserEndpoint", "type": "Endpoint", "versions": "0+",
              "about": "User-defined endpoint for Interactive Queries. " },
            { "name": "ClientTags", "type": "[]KeyValue", "versions": "0+",
              "about": "Used for rack-aware assignment algorithm." },
            { "name": "TaskOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog offsets for tasks." },
            { "name": "TaskEndOffsets", "type": "[]TaskOffset", "versions": "0+",
              "about": "Cumulative changelog end offsets for tasks." },

            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." }
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "name": "Endpoint", "versions": "0+", "fields": [
      { "name": "Host", "type": "string", "versions": "0+",
        "about": "host of the endpoint" },
      { "name": "Port", "type": "uint16", "versions": "0+",
        "about": "port of the endpoint" }
    ]},
    { "name": "TaskOffset", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partition", "type": "int32", "versions": "0+",
        "about": "The partition." },
      { "name": "Offset", "type": "int64", "versions": "0+",
        "about": "The offset." }
    ]},
    { "name": "TopicPartitions", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic ID." },
      { "name": "TopicName", "type": "string", "versions": "0+", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "Assignment", "versions": "0+", "fields": [
      { "name": "ActiveTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Active tasks for this client." },
      { "name": "StandbyTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Standby tasks for this client." },
      { "name": "WarmupTasks", "type": "[]TaskIds", "versions": "0+",
        "about": "Warm-up tasks for this client. " }
    ]},
    { "name": "TaskIds", "versions": "0+", "fields": [
      { "name": "SubtopologyId", "type": "string", "versions": "0+",
        "about": "The subtopology identifier." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions of the input topics processed by this member." }
    ]},
    { "name": "KeyValue", "versions": "0+", "fields": [
      { "name": "Key", "type": "string", "versions": "0+",
        "about": "key of the config" },
      { "name": "Value", "type": "string", "versions": "0+",
        "about": "value of the config" }
    ]},
    { "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": "ReplicationFactor", "type": "int16", "versions": "0+",
        "about": "The replication factor of the topic. Can be 0 if the default replication factor should be used." },
      { "name": "TopicConfigs", "type": "[]KeyValue", "versions": "0+",
        "about": "Topic-level configurations as key-value pairs."
      }
    ]}
  ]
}

...

Name	Type	Default	Doc
group.protocol	enum	classic	A flag which indicates if the new protocol should be used or not. It could be: classic or streams
topology.epoch	int	10	The epoch of the topology for the streams group. Ignored if `group.protocol=classic`

...

Space shortcuts

Child pages

Versions Compared

Old Version 70

New Version 71

Key

Topology updates