...

Share group membership is controlled by the group coordinator. Consumers in a share group use the heartbeat mechanism to join, leave and confirm continued membership of the share group, using the new ShareGroupHeartbeat RPC. Share-partition assignment is also piggybacked on the heartbeat mechanism. Share groups only support server-side assignors, which implement implements the org.apache.kafka.coordinator.group.shareassignor.SharePartitionAssignorShareGroupPartitionAssignor  interface. There is just one implementation of this interface so far, org.apache.kafka.coordinator.group.share.SimpleAssignor .

...

The share-group state data is not very amenable to log compaction. As a result, the share coordinator uses unlimited log retention and prunes the log records itself using ReplicaManager.deleteRecords. The share coordinator can delete all records before the latest ShareSnapshot for all active share-partitions. By taking periodic snapshots, the latest ShareSnapshot is replaced. For idle share-partitions, the share coordinator will periodically write a new ShareSnapshot so the older records can be pruned.

Using the share coordinator from the group coordinator

When a share-partition leader receives its first ShareFetch request for a topic-partition, it needs to initialize its The group coordinator is responsible for asking the share coordinator to initialize and delete the durable share-partition state. It finds the share coordinator using the FindCoordinator RPC using (key: "groupId:topicId:partition", key_type: SHARE ). Then, it sends the ReadShareGroupState RPC to the share coordinator. If the share coordinator has no share-partition state to return, it returns the UNKNOWN_TOPIC_OR_PARTITION error code indicating that this share-partition is not actually part of the share group. Otherwise, it returns the state to the share-partition leader which uses it to initialize and begin fetching records for the consumers. The SPSO returned might be -1 indicating that the initial SPSO needs to be set based on the group.share.auto.offset.reset configuration.

The share-partition leader must be aware of when the group coordinator is being used to alter the SPSO with a KafkaAdmin.alterShareGroupOffsets request. This only occurs when the group is empty. As a result, when the set of share sessions transitions from 0 to 1, the share-partition leader uses the ReadShareGroupOffsetsState RPC to validate its state epoch (this request is much cheaper for the share coordinator to handle than ReadShareGroupState ). We know that there are no acquired records, so re-initializing the share-partition leader is non-disruptive. If the state epoch has changed, the share-partition leader issues a ReadShareGroupState RPC to the share coordinator and uses the response to re-initialize.  

When a share-partition leader needs to update the durable share-partition state because of an acknowledgement or other state changed (such as a lock timeout), it sends the WriteShareGroupState RPC to the share coordinator. The share coordinator keeps track of the accumulated state of the share-partition and chooses how to record it to the share-state topic. Once it has successfully written to the topic and replication has completed, the RPC response is sent.

When a share partition is removed from a share group, perhaps because the topic is deleted or the administrator deletes the share-partition offsets (see KafkaAdmin.deleteShareGroupOffsets), the share-partition leader sends the DeleteShareGroupState RPC to the share coordinator. The share coordinator writes a final ShareSnapshot record with the special snapshot epoch value of -1. This acts as a deletion marker so the recovery processing of the share-group state topic sees the share-partition has been removed. There is no need to retain the deletion marker. Its only purpose is to make sure that a share coordinator reading records for a share-partition before it was removed, notices that those records apply to a defunct share-partition.

Examples with share-group state

Here are some examples showing the writing of share-group state.

The group coordinator uses the ShareGroupPartitionMetadata record to keep track of which share-partitions have been initialized.

The state is initialized in the following cases:

• When a topic is added to the set of subscribed topics for a share group and is not yet in the ShareGroupPartitionMetadata record
• When partitions are added to a topic which is already in the ShareGroupPartitionMetadata 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 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 with the "soft" state in the cluster metadata. This is how it observes relevant changes to topics, such as topic deletion. The ShareGroupPartitionMetadata contains the set of topic-partitions which are known to be initialized.

In order to add a topic, the group coordinator:

• 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 record with the topic added.

In order to add partitions to a known topic, the group coordinator:

• 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 record with the number of partitions for the topic increased.

In order to remove a topic, the group coordinator:

• 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 record with the topic removed.

Using the share coordinator from the share-partition leader

When a share-partition leader receives its first ShareFetch request for a topic-partition, it needs to initialize its share-partition state. It finds the share coordinator using the FindCoordinator RPC using (key: "groupId:topicId:partition", key_type: SHARE ). Then, it sends the ReadShareGroupState RPC to the share coordinator. If the share coordinator has no share-partition state to return, it returns the UNKNOWN_TOPIC_OR_PARTITION error code indicating that this share-partition is not actually part of the share group. Otherwise, it returns the state to the share-partition leader which uses it to initialize and begin fetching records for the consumers. The SPSO returned might be -1 indicating that the initial SPSO needs to be set based on the group.share.auto.offset.reset configuration.

The share-partition leader must be aware of when the group coordinator is being used to alter the SPSO with a KafkaAdmin.alterShareGroupOffsets request. This only occurs when the group is empty. As a result, when the set of share sessions transitions from 0 to 1, the share-partition leader uses the ReadShareGroupOffsetsState RPC to validate its state epoch (this request is much cheaper for the share coordinator to handle than ReadShareGroupState ). We know that there are no acquired records, so re-initializing the share-partition leader is non-disruptive. If the state epoch has changed, the share-partition leader issues a ReadShareGroupState RPC to the share coordinator and uses the response to re-initialize.  

When a share-partition leader needs to update the durable share-partition state because of an acknowledgement or other state changed (such as a lock timeout), it sends the WriteShareGroupState RPC to the share coordinator. The share coordinator keeps track of the accumulated state of the share-partition and chooses how to record it to the share-state topic. Once it has successfully written to the topic and replication has completed, the RPC response is sent.

When a share partition is removed from a share group, perhaps because the topic is deleted or the administrator deletes the share-partition offsets (see KafkaAdmin.deleteShareGroupOffsets), the share-partition leader sends the DeleteShareGroupState RPC to the share coordinator. The share coordinator writes a final ShareSnapshot record with the special snapshot epoch value of -1. This acts as a deletion marker so the recovery processing of the share-group state topic sees the share-partition has been removed. There is no need to retain the deletion marker. Its only purpose is to make sure that a share coordinator reading records for a share-partition before it was removed, notices that those records apply to a defunct share-partition.

Examples with share-group state

Here are some examples showing the writing of share-group state.

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": 100,
  "StateBatches": []
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": 110,
  "StateBatches": []
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": -1,
  "StateBatches": [
    {
      "FirstOffset": 110,
      "LastOffset": 110,
      "DeliveryState": 0 (Available),
      "DeliveryCount": 1
    }
  ]
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": -1,
  "StateBatches": [
    {
      "FirstOffset": 119,
      "LastOffset": 119,
      "DeliveryState": 2 (Acknowledged),
      "DeliveryCount": 1
    }
  ]
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": -1,
  "StateBatches": [
    {
      "FirstOffset": 111,
      "LastOffset": 112,
      "DeliveryState": 0 (Available),
      "DeliveryCount": 1
    }
  ]
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": -1,
  "StateBatches": [
    {
      "FirstOffset": 113,
      "LastOffset": 118,
      "DeliveryState": 2 (Acknowledged),
      "DeliveryCount": 1
    }
  ]
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": -1,
  "StateBatches": [
    {
      "FirstOffset": 110,
      "LastOffset": 110,
      "DeliveryState": 2 (Acknowledged),
      "DeliveryCount": 2
    }
  ]
}

{
  "GroupId": "G1",
  "TopicId": topic_uuid,
  "Partition": 0,
  "StartOffset": 120,
  "StateBatches": []
}

...

...

They are all subclasses of RetriableException .

Broker API

...

ShareGroupPartitionAssignor

The new org.apache.kafka.coordinator.group.shareassignor.SharePartitionAssignorShareGroupPartitionAssignor  interface is an interface implemented by server-side assignors for share groups. It signifies that the partition assignor is suitable for use with share groups.

package org.apache.kafka.coordinator.group.shareassignor;

import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.coordinator.group.assignor.AssignmentSpec;
import org.apache.kafka.coordinator.group.assignor.GroupAssignment;
import org.apache.kafka.coordinator.group.assignor.SubscribedTopicDescriber;

/**
 * Server side partition assignor used by the GroupCoordinator.
 *
 * The interface is kept in an internal module until KIP-932 is fully
 * implemented and ready to be released.
 */
@InterfaceStability.Unstable
public interface SharePartitionAssignorShareGroupPartitionAssignor {
extends PartitionAssignor   /**
     * Unique name for this assignor.
     */
    String name();

    /**
     * Assigns partitions to group members based on the given assignment specification and topic metadata.
     *
     * @param assignmentSpec           The assignment spec which includes member metadata.
     * @param subscribedTopicDescriber The topic and partition metadata describer.
     * @return The new assignment for the group.
     */
    GroupAssignment assign(
        AssignmentSpec assignmentSpec,
        SubscribedTopicDescriber subscribedTopicDescriber
    ) throws PartitionAssignorException;
}

Command-line tools

kafka-share-groups.sh

{
}

Command-line tools

kafka-share-groups.sh

A new tool called kafka-share-groups.sh is added for working with share groups. It has the following optionsA new tool called kafka-share-groups.sh is added for working with share groups. It has the following options:

...

This table gives the ACLs required for the new APIs.

Error codes

This KIP adds the following error codes the Kafka protocol.

...

{
  "type": "data",
  "name": "ShareGroupPartitionMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "InitializedTopics", "versions": "0+", "type": "[]TopicMetadata" },
     { "nameabout": "InitializingTopics", "versions": "0+", "type": "[]InitializingTopicMetadata" },The topics with initialized share-group state." }
  ],
  "commonStructs": [
    { "name": "DeletingTopicsTopicMetadata", "versions": "0+", "typefields": "[]TopicMetadata" }
  ],
  "commonStructs": [
    { "name": "TopicMetadata", "versions": "0+", "fields": [
      { "name[
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic identifier." },
      { "name": "TopicName", "type": "string", "versions": "0+",
        "about": "The topic name." },
      { "name": "NumPartitions", "type": "int32", "versions": "0+",
        "about": "The number of partitions." }
    ]}
  ]},
}

Share-group state

These records are written by the share coordinator on the __share_group_state  topic.

ShareSnapshotKey

{
  "type": "data",
 { "name": "InitializingTopicMetadataShareSnapshotKey",
  "versionsvalidVersions": "0+",
  "flexibleVersions": "none",
  "fields": [
      { "name": "TopicIdGroupId", "type": "uuidstring", "versions": "0+",
        "about": "The topicgroup identifierid.". },
      { "name": "TopicNameTopicId", "type": "stringuuid", "versions": "0+",
        "about": "The topic nameid." },
      { "name": "StartPartitionIndexPartition", "type": "int32", "versions": "0+",
      "The partition index." }
  ]
}

ShareSnapshotValue

{
  "abouttype": "data"The,
 start partition index of the partitions being initialized." },
  "name": "ShareSnapshotValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "EndPartitionIndexSnapshotEpoch", "type": "int32uint16", "versions": "0+",
       "about": "The end partition index of the partitions being initializedsnapshot epoch." },
    ]}
  ]
}

The InitializingTopics  field is used as the first stage of a two-stage process to initialize the persistent state for a set of share-partitions. When partitions are being initialized, StartPartitionIndex  is the index of the first partition being initialized and EndPartitionIndex  is the index of the last partition being initialized, such as when the number of partitions for a topic is increased. Once the share coordinator successfully responds to InitializeShareGroupState  , the topic-partitions are moved into the InitializedTopics  field. A topic can of course be in both the InitializedTopics  and the InitializingTopics  field when its number of partitions has been increased.

In a similar way, the DeletingTopics  field is used as the first stage of a two-stage process to delete the persistent state for a set of share-partitions.

Share-group state

These records are written by the share coordinator on the __share_group_state  topic.

ShareSnapshotKey

{
 { "name": "StateEpoch", "type": "dataint32",
  "nameversions": "ShareSnapshotKey0+",
      "validVersionsabout": "0",
The state "flexibleVersions": "none",
  "fields": [epoch for this share-partition." },
     { "name": "GroupIdStartOffset", "type": "stringint64", "versions": "0",
      "about": "The share-partition groupstart idoffset." },
    { "name": "TopicIdStateBatches", "type": "uuid[]StateBatch", "versions": "0",
      "aboutfields": "The topic id." },
[
      { "name": "PartitionFirstOffset", "type": "int32int64", "versions": "0",
      "The partition index." }
  ]
}

ShareSnapshotValue

{
  "type"about": "data",
The first "name": "ShareSnapshotValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
offset of this state batch." },
      { "name": "SnapshotEpochLastOffset", "type": "uint16int64", "versions": "0",
        "about": "The snapshot epoch last offset of this state batch." },
      { "name": "StateEpochDeliveryState", "type": "int32int8", "versions": "0+",
        "about": "The delivery state epoch for this share-partition.- 0:Available,2:Acked,4:Archived" },
       { "name": "StartOffsetDeliveryCount", "type": "int64int16", "versions": "0",
        "about": "The share-partition start offsetdelivery count." },
    ]} 
  ]
}

ShareUpdateKey

{
  "type": "data",
  "name": "StateBatchesShareUpdateKey",
  "typevalidVersions": "[]StateBatch1",
  "versionsflexibleVersions": "0none",
  "fields": [
      { "name": "FirstOffsetGroupId", "type": "int64string", "versions": "0",
        "about": "The first offset of this state batchgroup id." },
      { "name": "LastOffsetTopicId", "type": "int64uuid", "versions": "0",
        "about": "The last offset of this state batchtopic id." },
      { "name": "DeliveryStatePartition", "type": "int8int32", "versions": "0",
        "about": "The delivery state - 0:Available,2:Acked,4:Archived" },
      { "name": "DeliveryCount", "type": "int16", "versions": "0",
        "about": "The delivery count." }
    ]} partition index." }
  ]
}

...

ShareUpdateValue

{
  "type": "data",
  "name": "ShareUpdateKeyShareUpdateValue",
  "validVersions": "10",
  "flexibleVersions": "none0+",
  "fields": [
    { "name": "GroupIdSnapshotEpoch", "type": "stringuint16", "versions": "0",
      "about": "The groupsnapshot idepoch." },
    { "name": "TopicIdStartOffset", "type": "uuidint64", "versions": "0",
      "about": "The topic id share-partition start offset, or -1 if the start offset is not being updated." },
    { "name": "PartitionStateBatches", "type": "int32[]StateBatch", "versions": "0",
      "The partition index." }
  ]
}

ShareUpdateValue

{
  "type"fields": [
      { "name": "dataFirstOffset",
  "nametype": "ShareUpdateValueint64",
  "validVersionsversions": "0",
        "flexibleVersionsabout": "0+"The first offset of this state batch." },
  "fields": [
    { "name": "SnapshotEpochLastOffset", "type": "uint16int64", "versions": "0",
        "about": "The snapshot epoch last offset of this state batch." },
      { "name": "StartOffsetDeliveryState", "type": "int64int8", "versions": "0",
        "about": "The share-partition start offset, or delivery state -1 if the start offset is not being updated." },
    { "name": "StateBatches", "type": "[]StateBatch", "versions": "0", "fields": [ 0:Available,2:Acked,4:Archived" },
      { "name": "FirstOffsetDeliveryCount", "type": "int64int16", "versions": "0",
        "about": "The first offset of this state batchdelivery count." },
    ]}  { "name": "LastOffset", "type": "int64", "versions": "0",
        "about": "The last offset of this state batch." },
      { "name": "DeliveryState", "type": "int8", "versions": "0",
        "about": "The delivery state - 0:Available,2:Acked,4:Archived" },
      { "name": "DeliveryCount", "type": "int16", "versions": "0",
        "about": "The delivery count." }
    ]} 
  ]
}

Metrics

Broker Metrics

The following new broker metrics should be added:

  ]
}

Metrics

Broker Metrics

The following new broker metrics should be added:

Client metrics

The following new client metrics should be added:

Future Work

There are some obvious extensions to this idea which are not included in this KIP in order to keep the scope manageable.

...