Page History

...

Whenever the group epoch is larger than the target assignment epoch, the group coordinator triggers the computation of a new target assignment based on the latest group metadata using a server-side assignor. For a share group, the group coordinator does not persist the assignment. The assignment epoch becomes the group epoch of the group metadata used to compute the assignment.

...

For a share group, the group coordinator persists three seven kinds of records:

• ShareGroupMetadata - this is written to reserve the group's ID as a share group in the namespace of groups.
• ShareGroupMemberMetadata - this is written to allow group membership to continue seamlessly across a group coordinator change.
• ShareGroupPartitionMetadata - this is written to persist the metadata about the partitions the group is assigning to the members.
• ShareGroupTargetAssignmentMetadata - this is written to persist the assignment epoch.
• ShareGroupTargetAssignmentMember - this is written to persist the target assignment for a member.
• ShareGroupCurrentMemberAssignment - this is written to persist the current assignment for a member, and also to maintain the sequence of member epochs.
• ShareGroupStatePartitionMetadata - this is written whenever the set of topic-partitions being consumed in the share group changes. Its purpose is to keep track of which topic-partitions will have share-group persistent state.

When the group coordinator fails over, the newly elected coordinator replays the state from the __consumer_offsets  partition. This means a share group will remain in existence across the fail-over, along with the list of members . However, the assignments are not persisted and will be recalculated by the new group coordinatorand their assignments. It will bump the group epoch as a result.

...

The group coordinator is responsible for asking the share coordinator to initialize and delete the durable share-partition state. The group coordinator uses the ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record to keep track of which share-partitions have been initialized.

...

• When a topic is added to the set of subscribed topics for a share group and is not yet in the ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record
• When partitions are added to a topic which is already in the ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record
• When KafkaAdmin.alterShareGroupOffsets  is used to reset the SPSO for a share-partition

The state is deleted in the following cases:

• When a topic in the ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record is deleted
• When KafkaAdmin.deleteShareGroupOffsets  is used to delete state for a share-partition, most likely as a result of an administrator cleaning up a topic which is no longer in use by this share group
• When the share group is deleted

The group coordinator periodically reconciles its "hard" state in the ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata with the "soft" state in the cluster metadata. This is how it observes relevant changes to topics, such as topic deletion. The ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata contains the set of topic-partitions which are known to be initialized.

...

• Sends the InitializeShareGroupState  RPC to the share coordinators for all of the partitions of the topic. This step can be repeated to handle failures, such as group coordinator failover.
• When it has successful responses for all partitions, it writes a ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record with the topic added.

...

• Sends the InitializeShareGroupState  RPC to the share coordinators for all of the new partitions of the topic. This step can be repeated to handle failures, such as group coordinator failover.
• When it has successful responses for all partitions, it writes a ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record with the array of partitions for the topic increased.

...

• If the topic still exists, it writes a ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record with the topic added to the DeletingTopics  array. This enables group coordinator failover to continue the deletion. If the topic does not exist, the requirement to continue deletion can be inferred by the non-existence of the topic.
• Sends the DeleteShareGroupState  RPC to the share coordinators for all of the partitions of the topic. This step can be repeated to handle failures, such as group coordinator failover.
• When it has successful responses for all partitions, it writes a ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record with the topic removed.

...

...

Group configuration

The following dynamic group configuration properties are added. These are properties for which it would be problematic to have consumers in the same share group using different behavior if the properties were specified in the consumer clients themselves.

Group configuration

The following dynamic group configuration properties are added. These are properties for which it would be problematic to have consumers in the same share group using different behavior if the properties were specified in the consumer clients themselves.

...

This section describes the new record types.

Group and assignment metadata

In order to coexist properly with consumer groups, the group metadata records for share groups are persisted by the group coordinator to the compacted __consumer_offsets  topic.

With the exception of the ShareGroupStatePartitionMetadata record, all of the record types are analogous to those introduced in KIP-848 for consumer groups.

For each share group, a ShareGroupMetadata record is written for the group epoch, and a ShareGroupPartitionMetadata record is written for the topic-partitions being assigned by the group. For each share group, a single ShareGroupMetadata record is written. When the group is deleted, a tombstone record is records are written.

For each member, there is a ShareGroupMemberMetadata record which enables group membership to continue seamlessly across a group coordinator change. When the member leaves, a tombstone record is written.

For each share group, a ShareGroupTargetAssignmentMetadata record is written to record the group epoch used to compute the assignment. For each member, there is a ShareGroupTargetAssignmentMember record which persists the target assignment, and a ShareGroupCurrentMemberAssignment record which persists the current assignment and is also used to keep track of the member epoch.

There is also a ShareGroupPartitionMetadata ShareGroupStatePartitionMetadata record which contains a list of all share-partitions which have been assigned in the share group. It is used to keep track of which share-partitions have persistent state.

...

{
  "type": "data",
  "name": "ShareGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." }
  ]
}

...

ShareGroupPartitionMetadataKey

{
  "type": "data",
  "name": "ShareGroupMemberMetadataKeyShareGroupPartitionMetadataKey",
  "validVersions": "109",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "9",
      "about": "The group id." }
  ]
}

ShareGroupPartitionMetadataValue

{
  "type": "data",
  "name": "ShareGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "versions": "0+", "type": "10[]TopicMetadata",
      "about": "The list of topic metadata.", "fields": [
      { "name": "TopicId", "versions": "0+", "type": "uuid",
        "about": "The topic id." },
      { "name": "TopicName", "versions": "0+", "type": "string",
        "about": "The topic name." },
      { "name": "NumPartitions", "versions": "0+", "type": "int32",
        "about": "The number of partitions of the topic." },
      { "name": "PartitionMetadata", "versions": "0+", "type": "[]PartitionMetadata",
        "about": "Partitions mapped to a set of racks. If the rack information is unavailable for all the partitions, an empty list is stored", "fields": [
        { "name": "Partition", "versions": "0+", "type": "int32",
          "about": "The partition number." },
        { "name": "Racks", "versions": "0+", "type": "[]string",
          "about": "The set of racks that the partition is mapped to." }
      ]}
    ]}
  ]
}

ShareGroupMemberMetadataKey

{
  "type": "data",
  "name": "ShareGroupMemberMetadataKey",
  "validVersions": "10",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "10",
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": "10",
      "about": "The member id." }
  ]
}

ShareGroupMemberMetadataValue

{
  "type": "data",
  "name": "ShareGroupMemberMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "RackId", "versions": "0+", "nullableVersions": "0+", "type": "string",
      "about": "The (optional) rack id." },
    { "name": "ClientId", "versions": "0+", "type": "string",
      "about": "The client id." },
    { "name": "ClientHost", "versions": "0+", "type": "string",
      "about": "The client host." },
    { "name": "SubscribedTopicNames", "versions": "0+", "type": "[]string",
      "about": "The list of subscribed topic names." }
  ]
}

ShareGroupTargetAssignmentMetadataKey

{
  "type": "data",
  "name": "ShareGroupTargetAssignmentMetadataKey",
  "validVersions": "12",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "12",
      "about": "The group id." }
  ]
}

ShareGroupTargerAssignmentMetadataValue

{
  "type": "data",
  "name": "ShareGroupTargetAssignmentMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "AssignmentEpoch", "versions": "0+", "type": "int32",
      "about": "The assignment epoch." }
  ]
}

ShareGroupTargetAssignmentMemberKey

{
  "type": "data",
  "name": "ShareGroupTargetAssignmentMemberKey",
  "validVersions": "13",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "13",
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": "13",
      "about": "The member id." }
  ]
}

ShareGroupTargetAssignmentMemberValue

{
  "type": "data",
  "name": "ShareGroupTargetAssignmentMemberValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicPartitions", "versions": "0+", "type": "[]TopicPartition",
      "about": "The assigned partitions.", "fields": [
      { "name": "TopicId", "versions": "0+", "type": "uuid" },
      { "name": "Partitions", "versions": "0+", "type": "[]int32" }
    ]}
  ]
}

ShareGroupCurrentMemberAssignmentKey

{
  "type": "data",
  "name": "ShareGroupCurrentMemberAssignmentKey",
  "validVersions": "14",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "14",
      "about": "The group id." },
    { "name": "MemberId", "type": "string", "versions": "14",
      "about": "The member id." }
  ]
}

ShareGroupCurrentMemberAssignmentValue

{
  "type": "data",
  "name": "ShareGroupCurrentMemberAssignmentValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MemberEpoch", "versions": "0+", "type": "int32",
      "about": "The current member epoch that is expected from the member in the heartbeat requestThe group id." },
    { "name": "MemberIdPreviousMemberEpoch", "typeversions": "string0+", "versionstype": "10int32",
      "about": "The member idIf the last epoch bump is lost before reaching the member, the member will retry with the previous epoch." },
    ]
}

ShareGroupMemberMetadataValue

{
  "typename": "dataState",
  "nameversions": "ShareGroupMemberMetadataValue0+",
  "validVersionstype": "0int8",
      "flexibleVersionsabout": "0+",
  "fields": [The member state. See MemberState for the possible values." },
    { "name": "RackIdAssignedPartitions", "versions": "0+", "nullableVersions": "0+", "type": "string[]TopicPartitions",
      "about": "The (optionalpartitions assigned to (or owned by) rackthis idmember." }
  ],
  "commonStructs": [
    { "name": "ClientIdTopicPartitions", "versions": "0+", "typefields": "string",[
      { "aboutname": "The client id." },
    { "nameTopicId", "type": "ClientHostuuid", "versions": "0+",
 "type": "string",
      "about": "The clienttopic hostId." },
      { "name": "SubscribedTopicNamesPartitions", "versionstype": "0+[]int32", "typeversions": "[]string0+",
        "about": "The partition list of subscribed topic names." Ids." }
    ]}
  ]
}

...

ShareGroupStatePartitionMetadataKey

{
  "type": "data",
  "name": "ShareGroupPartitionMetadataKeyShareGroupStatePartitionMetadataKey",
  "validVersions": "915",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "915",
      "about": "The group id." }
  ]
}

...

ShareGroupStatePartitionMetadataValue

{
  "type": "data",
  "name": "ShareGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "InitializedTopics", "versions": "0+", "type": "[]TopicPartitionsInfo",
      "about": "The topics with initialized share-group state." },
    { "name": "DeletingTopics", "versions": "0+", "type": "[]TopicInfo",
      "about": "The topics whose share-group state is being deleted." }
  ],
  "commonStructs": [
    { "name": "TopicPartitionsInfo", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic identifier." },
      { "name": "TopicName", "type": "string", "versions": "0+",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "TopicInfo", "versions": "0+", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic identifier." },
      { "name": "TopicName", "type": "string", "versions": "0+",
        "about": "The topic name." }
    ]}
  ]
}

...

{
  "type": "data",
  "name": "ShareUpdateKey",
  "validVersions": "1",
  "flexibleVersions": "none",
  "fields": [
    { "name": "GroupId", "type": "string", "versions": "01",
      "about": "The group id." },
    { "name": "TopicId", "type": "uuid", "versions": "01",
      "about": "The topic id." },
    { "name": "Partition", "type": "int32", "versions": "01",
      "about": "The partition index." }
  ]
}

...