It is generated by the
client once and must be used during its lifetime.

Table of Contents

maxLevel	4

Status

Current state: Under Discussion Accepted

Discussion thread: Thread 1 and Thread 2here

JIRA: here

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

When a client side assignor is used, the group coordinator requests the assignment from one group member by notifying him it via the heartbeat protocol. The chosen member uses the ConsumerGroupPrepareAssignment API and the ConsumerGroupInstallAssignment API to respectively get the current state of the group and to install the computed assignment. Thanks to this, the input of the client side assignor is entirely driven by the group coordinator. The consumer is no longer responsible for maintaining any state besides its assigned partitions.

...

We will support two assignors out of the box for Apache Kafka:

range - org.apache.kafka.server.group.consumer.RangeAssignor - An assignor which co-partitions topics.
uniform - org.apache.kafka.server.group.consumer.UniformAssignor - An assignor which uniformly assign partitions amongst the members. This is somewhat similar to the existing "sticky" assignor.

...

The member uses the ConsumerGroupHeartbeat API to establish a session between him and the with the group coordinator. The member is expected to heartbeat every group.consumer.heartbeat.interval.ms in order to keep its session opened. If it does not heartbeat at least once within the group.consumer.session.timeout.ms, the group coordinator will kick him the member out from the group. group.consumer.heartbeat.interval.ms is defined on the server side and the member is told about it in the heartbeat response. The group.consumer.session.timeout.ms is also defined on the server side.

...

The member joins the group by sending an heartbeat with no Member ID and a member epoch equals to 0. He can rejoins the group with a with a member epoch equals to 0. He can leaves the group by using a member epoch equals to -1. The member must commit its offsets before leaving.

Fencing

The group coordinator ensures that requests comes from a known Member ID. Any request is rejected with the UNKNOWN_MEMBER_ID error otherwise. It also ensures that the Member Epoch matches the expected member epoch. If not, the request is rejected with the FENCED_MEMBER_EPOCH error. In this case, the member is expected to immediately gives up all its partitions and rejoin the group with the same member ID and a member epoch equals to zero. Details for every API are given in the Public Interfaces section.

...

Static membership, introduced in KIP-345, is still supported by this new rebalance protocol. When a member wants to leave temporary – e.g. while being bounced – it should send an heartbeat with a member epoch equals to -2. This signals to the group coordinator that the member left but will rejoin within the session timeout. When the member rejoins with the same Instance instance ID, the group coordinator replaces the old member with by the new member . The new member can continue from where it left offand gives back its current assignment.

If the leaving member does not rejoin within the session timeout, the group coordinator kicks it out from the group. If a new member joins with an existing instance ID before the previous member left, the new member is rejected with a UNRELEASED_INSTANCE_ID error as long as the previous member is still present.

Consumer Group States

EMPTY

...

The group configurations are stored in the controller like all the other dynamic configurations in the cluster. This allows configurations to be installed independently from wether whether the group exists or not. Configurations are also preserved if the group is deleted. In ZK mode, the dynamic group configurations will be store in the /config/groups znode. In KRaft mode, they are stored in the ConfigRecord.

...

It has to setup the session timeouts for all the members (like today).
It has to check wether whether the topic-partition metadata has changed and potentially trigger a rebalance for the group if it has.
It has to check wether whether new topics match the regex subscriptions and trigger a rebalance for the group if new topic do.
It has to check wether whether a new assignment is required for the group (group epoch != assignment epoch). If it is the case, the group coordinator can directly compute it using the server side assignor or can trigger a client side assignment computation.

...

When the member has got a new target assignment, the group coordinator will notify him it that it has to revoke partitions if any partitions must be revoked. The member can determine the revoked partitions by computing the difference between its current local assignment and the assignment received from the group coordinator. The consumer stops fetching from those revoked partitions and, If auto commit is enabled, commits their offsets. Finally, it calls ConsumerRebalanceListener#onPartitionsRevoked and inform the group coordinator that the revocation process in its next heartbeat.

...

The idea is to manage each members individually while relying on the new engine to do the synchronization between them. Each member will use rebalance loops to update the group coordinator and collect their assignment. The group coordinator will ensure that a rebalance is triggered when one needs to update his its assignments. This process is detailed in the next sub-chapters:

...

FENCED_MEMBER_EPOCH - The member epoch is fenced by the coordinator. The member must abandon all its partitions and rejoins.
STALE_MEMBER_EPOCH - The member epoch is stale. The member must retry after receiving its updated member epoch via the ConsumerGroupHeartbeat API.
UNRELEASED_INSTANCE_ID - The instance ID is still used by another member. The member must leave first.
UNSUPPORTED_ASSIGNOR - The assignor used by the member or its version range are not supported by the group.

...

Code Block

linenumbers	true

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["zkBroker", "broker"],
  "name": "ConsumerGroupHeartbeatRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
        "about": "The group identifier." },
      { "name": "MemberId", "type": "string", "versions": "0+",
        "about": "The member id generated by the servercoordinator. The member id must be kept during the entire lifetime of the member." },
      { "name": "MemberEpoch", "type": "int32", "versions": "0+",
        "about": "The current member epoch; 0 to join the group; -1 to leave the group." },
      { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if not provided or if it didn't change since the last heartbeat; the instance Id otherwise." },
      { "name": "RackId", "type": "string", "versions": "0+",  "nullableVersions": "0+", "default": "null",
        "about": "null if not provided or if it didn't change since the last heartbeat; the rack ID of consumer otherwise." },
      { "name": "RebalanceTimeoutMs", "type": "int32", "versions": "0+", "default": -1,
        "about": "-1 if it didn't chance since the last heartbeat; the maximum time in milliseconds that the coordinator will wait on the member to revoke its partitions otherwise." },
      { "name": "SubscribedTopicNames", "type": "[]string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if it didn't change since the last heartbeat; the subscribed topic names otherwise." },
      { "name": "SubscribedTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if it didn't change since the last heartbeat; the subscribed topic regex otherwise" },
      { "name": "ServerAssignor", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if not used or if it didn't change since the last heartbeat; the server side assignor to use otherwise." }, 
      { "name": "ClientAssignors", "type": "[]Assignor", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if not used or if it didn't change since the last heartbeat; the list of client-side assignors otherwise.", "fields": [
          { "name": "Name", "type": "string", "versions": "0+",
            "about": "The name of the assignor." },
          { "name": "MinimumVersion", "type": "int16", "versions": "0+",
            "about": "The minimum supported version for the metadata." },
          { "name": "MaximumVersion", "type": "int16", "versions": "0+",
            "about": "The maximum supported version for the metadata." },
          { "name": "Reason", "type": "int8", "versions": "0+",
            "about": "The reason of the metadata update." }, 
          { "name": "MetadataVersion", "type": "int16", "versions": "0+",
            "about": "The version of the metadata." },
          { "name": "MetadataBytes", "type": "bytes", "versions": "0+",
            "about": "The metadata." }
      ]},
      { "name": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "null if it didn't change since the last heartbeat; the partitions owned by the member.", "fields": [
          { "name": "TopicId", "type": "uuid", "versions": "0+",
            "about": "The topic ID." },
          { "name": "Partitions", "type": "[]int32", "versions": "0+",
            "about": "The partitions." }
      ]}
  ]
}

...

Lookups the group or creates it.
Creates the member should the member epoch be zero or checks whether it exists. If it does not exist, UNKNOWN_MEMBER_ID is returned.
Checks wether 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 he knows about and his its request will be rejected with a FENCED_MEMBER_EPOCH. This is not optimal. Instead, the group coordinator could accept the request if the partitions owned by the members are a subset of the target partitions. If it is the case, it is safe to transition the member to its target epoch again.
Updates the members informations if any. The group epoch is incremented if there is any change. See "Group Epoch - Trigger a rebalance" chapter for details about the rebalance triggers.
Reconcile the member assignments as explained earlier in this document.

...

Code Block

language	js
linenumbers	true

{
  "apiKey": TBD,
  "type": "response",
  "name": "ConsumerGroupHeartbeatResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED
  // - NOT_COORDINATOR
  // - COORDINATOR_NOT_AVAILABLE
  // - COORDINATOR_LOAD_IN_PROGRESS
  // - INVALID_REQUEST
  // - UNKNOWN_MEMBER_ID
  // - FENCED_MEMBER_EPOCH
  // - UNSUPPORTED_ASSIGNOR
  // - UNRELEASED_INSTANCE_ID
  // - GROUP_MAX_SIZE_REACHED
  "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": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The top-level error code, 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": "MemberId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The member id generated by the coordinator. Only provided when the member joins with MemberEpoch == 0." }, 
    { "name": "MemberEpoch", "type": "int32", "versions": "0+",
      "about": "The member epoch." },
    { "name": "ShouldComputeAssignment", "type": "bool", "versions": "0+",
      "about": "True if the member should compute the assignment for the group." },
    { "name": "HeartbeatIntervalMs", "type": "int32", "versions": "0+",
      "about": "The heartbeat interval in milliseconds." }, 
    { "name": "Assignment", "type": "Assignment", "versions": "0+", "nullableVersions": "0+", "default": "null",
	  "about": "null if not provided; the assignment otherwise.", "fields": [
        { "name": "Error", "type": "int8", "versions": "0+",
          "about": "The assigned error." },
        { "name": "AssignedTopicPartitions", "type": "[]TopicPartitions", "versions": "0+",
          "about": "The partitions assigned to the member that can be used immediately." },
        { "name": "PendingTopicPartitions", "type": "[]TopicPartitions", "versions": "0+",
          "about": "The partitions assigned to the member that cannot be used because they are not released by their former owners yet." },
        { "name": "MetadataVersion", "type": "int16", "versions": "0+",
          "about": "The version of the metadata." },
        { "name": "MetadataBytes", "type": "bytes", "versions": "0+",
          "about": "The assigned metadata." }
	]}
  ],
  "commonStructs": [
    { "name": "TopicPartitions", "versions": "0+", "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+",
          "about": "The topic ID." },
        { "name": "Partitions", "type": "[]int32", "versions": "0+",
          "about": "The partitions." }
    ]}
  ]
}

...

Upon receiving the UNKNOWN_MEMBER_ID or FENCED_MEMBER_EPOCH error, the consumer abandon all its partitions and rejoins with the same member id and the epoch 0.

Upon receiving the UNRELEASED_INSTANCE_ID error, the consumer should fail.

ConsumerGroupPrepareAssignment API

...

When the group coordinator handle a ConsumerGroupPrepareAssignmentRequest request:

Checks wether whether the group exists. If it does not, GROUP_ID_NOT_FOUND is returned.
Checks wether whether the member exists. If it does not, UNKNOWN_MEMBER_ID is returned.
Checks wether whether the member epoch matches the current member epoch. If it does not, STALE_MEMBER_EPOCH is returned.
Checks wether whether the member is the chosen one to compute the assignment. If it does not, UNKNOWN_MEMBER_ID is returned.
Returns the group state of the group.

...

When the group coordinator handle a ConsumerGroupInstallAssignmentRequest request:

Checks wether whether the group exists. If it does not, GROUP_ID_NOT_FOUND is returned.
Checks wether whether the member exists. If it does not, UNKNOWN_MEMBER_ID is returned.
Checks wether whether the member epoch matches the current member epoch. If it does not, STALE_MEMBER_EPOCH is returned.
Checks wether whether the member is the chosen one to compute the assignment. If it does not, UNKNOWN_MEMBER_ID is returned.
Validates the assignment based on the information used to compute it. If it is not valid, INVALID_ASSIGNMENT is returned.
Installs the new target assignment.

...

When the group coordinator handle a ConsumerGroupDescribeRequest request:

Checks wether whether the group ids exists. If it does not, GROUP_ID_NOT_FOUND is returned.
Looks up the groups and returns the response.

...

When a member is deleted, a tombstone for him it is written to the partition.

...

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupPartitionMetadataValue",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
        { "name": "Epoch", "versions": "0+", "type": "int32" },
        { "name": "TopicPartitionMetadataTopics", "versions": "0+",
          "type": "[]TopicPartitionTopicMetadata", "fields": [
            { "name": "TopicId", "versions": "0+", "type": "uuid" },
            { "name": "NumPartitions", "versions": "0+", "type": "int32" }
          ]}
    ], 
}

...

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupMemberMetadataValue",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
        { "name": "GroupEpoch", "versions": "0+", "type": "int32" },
        { "name": "InstanceId", "versions": "0+", "nullableVersions": "0+", "type": "string" },
        { "name": "RackId", "versions": "0+", "nullableVersions": "0+", "type": "string" },
        { "name": "ClientId", "versions": "0+", "type": "string" },
        { "name": "ClientHost", "versions": "0+", "type": "string" },
        { "name": "SubscribedTopicNames", "versions": "0+", "type": "[]string" },
        { "name": "SubscribedTopicRegex", "versions": "0+", "type": "string" },
        { "name": "Assignors", "versions": "0+",
          "type": "[]Assignor", "fields": [
            { "name": "Name", "versions": "0+", "type": "string" },
            { "name": "MinimumVersion", "versions": "0+", "type": "int16" },
            { "name": "MaximumVersion", "versions": "0+", "type": "int16" },
            { "name": "Reason", "versions": "0+", "type": "int8" },
			{ "name": "Version", "versions": "0+", "type": "int16" },
            { "name": "Metadata", "versions": "0+", "type": "bytes" }
          ]}
    ], 
}

...

The target assignment is stored in N + 1 records where N is the number of members in the group. The records for the members are written first and followed by the assignment metadata. When a single record.

...

new assignment is computed, the group coordinator will compare it with the current assignment and only write the difference between the two assignments to the __consumer_offsets partition. The assignment must be atomic so the group coordinator will ensure that all the records are written in a single batch. This limit the size of the batch to 1MB (the default value). Given the incremental nature of the protocol, 1MB should be sufficient in most case here.

ConsumerGroupTargetAssignmentMetadataKey

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupTargetAssignmentKeyConsumerGroupTargetAssignmentMetadataKey",
    "validVersions": "6",
    "flexibleVersions": "none",
    "fields": [
      	{ "name": "GroupId", "type": "string", "versions": "6" }
    ]
}

...

ConsumerGroupTargetAssignmentMetadataValue

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupTargetAssignmentValueConsumerGroupTargetAssignmentMetadataValue",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
        { "name": "AssignmentEpoch", "versions": "0+", "type": "int32" },
        { "name": "Members", "versions": "0+", "type": "[]Member", "fields": [
        	{ "name": "MemberId", "versions": "0+", "type": "string" },
                ]
}

The AssignmentEpoch corresponds to the group epoch used to compute the assignment. It is not necessarily the most recent group epoch because the assignment is computed asynchronously when a client-side assignor is used. When a client side assignor is used, the assignment is computed asynchronously. While it is computed for the group at epoch X, the group may have already advanced to epoch X+1 due to another event (e.g. new member joined). In this case, we have chosen to install the assignment computed for epoch X and to trigger a new assignment computation right away.

ConsumerGroupTargetAssignmentMemberKey

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupTargetAssignmentMemberKey",
    "validVersions": "7",
    "flexibleVersions": "none",
    "fields": [
      	{ "name": "ErrorGroupId", "versionstype": "0+string", "typeversions": "int87" },
             { "name": "TopicPartitionsMemberId", "versionstype": "0+string",
          	  "typeversions": "7" }
     ]
}

ConsumerGroupTargetAssignmentMemberValue

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupTargetAssignmentMemberValue",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
        { "name": "Error", "versions": "0+", "type": "int8" },
        { "name": "TopicPartitions", "versions": "0+",
       	  "type": "[]"[]TopicPartition", "fields": [
            	      	  { "name": "TopicId", "versions": "0+", "type": "uuid" },
            	          { "name": "Partitions", "versions": "0+", "type": "[]int32" }
        	]},
        	{ "name": "VersionMetadataVersion", "versions": "0+", "type": "int16" },
        	{ "name": "MetadataMetadataBytes", "versions": "0+", "type": "bytes" }
        ]
    ]
}

Current Member Assignment

The AssignmentEpoch corresponds to the group epoch used to compute the assignment. It is not necessarily the last one.

Current Member Assignment

The current member assignment represents, as the name current member assignment represents, as the name suggests, the current assignment of a given member.

When a member is deleted from the group, a tombstone for him it is written to the partition.

...

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupCurrentMemberAssignmentKey",
    "validVersions": "78",
    "flexibleVersions": "none",
    "fields": [
      	{ "name": "GroupId", "type": "string", "versions": "78" },
      	{ "name": "MemberId", "type": "string", "versions": "78" },
    ]
}

ConsumerGroupCurrentMemberAssignmentValue

Code Block

language	js
linenumbers	true

{
    "type": "data",
    "name": "ConsumerGroupCurrentMemberAssignmentValue",
    "validVersions": "0",
    "flexibleVersions": "0+",
    "fields": [
        { "name": "MemberEpoch", "versions": "0+", "type": "int32" },
		{ "name": "Error", "versions": "0+", "type": "int8" },
        { "name": "TopicPartitions", "versions": "0+",
          "type": "[]TopicPartition", "fields": [
            { "name": "TopicId", "versions": "0+", "type": "uuid" },
            { "name": "Partitions", "versions": "0+", "type": "[]int32" }
        ]},
        { "name": "VersionMetadataVersion", "versions": "0+", "type": "int16" },
        { "name": "MetadataMetadataBytes", "versions": "0+", "type": "bytes" }
    ], 
}

...

Offsets

OffsetCommitValue

Code Block

language	js
linenumbers	true

{
  "type": "data",
  "name": "OffsetCommitValue",
  "validVersions": "0-4",
  "flexibleVersions": "4+",
  "fields": [
    { "name": "offset", "type": "int64", "versions": "0+" },
    { "name": "leaderEpoch", "type": "int32", "versions": "3+", "default": -1, "ignorable": true },
    { "name": "metadata", "type": "string", "versions": "0+" },
    { "name": "commitTimestamp", "type": "int64", "versions": "0+" },
    { "name": "expireTimestamp", "type": "int64", "versions": "1", "default": -1, "ignorable": true },
    // Adds TopicId field.
    { "name": "topicId", "type": "uuid", "versions": "4", "ignorable": true }
  ]
}

...

Code Block

language	java
linenumbers	true

package org.apache.kafka.server.group.consumer;

public interface PartitionAssignor {

    class AssignmentSpec {
        /**
         * The members keyed by member id.
         */
        Map<String, AssignmentMemberSpec> members;

        /**
         * The topics' metadata keyed by topic id
         */
        Map<Uuid, AssignmentTopicMetadata> topics;
    }

    class AssignmentMemberSpec {
        /**
         * The instance ID if provided.
         */
        Optional<String> instanceId;

        /**
         * The rack ID if provided.
         */
        Optional<String> rackId;

        /**
         * The topics that the member is subscribed to.
         */
        Collection<String> subscribedTopics;

        /**
         * The current target partitions of the member.
         */
        Collection<TopicPartition> targetPartitions;
    }

    class AssignmentTopicMetadata {
        /**
         * The topic name.
         */
        String topicName;

        /**
		 * The number of partitions.
		 */
		int numPartitions; 
    }

    class GroupAssignment {
        /**
         * The member assignments keyed by member id.
         */
        Map<String, MemberAssignment> members;
    }

    class MemberAssignment {
        /**
         * The target partitions assigned to this member.
         */
        Collection<TopicPartition> targetPartitions;
    }

    /**
     * Unique name for this assignor.
     */
    String name();

    /**
     * Perform the group assignment given the current members and
     * topic metadata.
     *
     * @param assignmentSpec The assignment spec.
     * @return The new assignment for the group.
     */
    GroupAssignment assign(AssignmentSpec assignmentSpec) throws PartitionAssignorException;
}

Broker Metrics

The set of new metrics is not clear at the moment. We plan to amend the KIP later on when progress on the implementation would have been made.

Existing generic group metrics have been migrated, with the same metric names except for
NumGroups which reported the number of generic groups. This metric changed to
kafka.server:type=group-coordinator-metrics,name=group-count,protocol={consumer|generic}
number of groups based on type where type is the rebalance protocol
kafka.server:type=group-coordinator-metrics,name=partition-count,state={loading|active|failed}
number of __consumer_offsets partitions based on state
kafka.server:type=group-coordinator-metrics,name=event-queue-size
event accumulator queue size
kafka.server:type=group-coordinator-metrics,name=consumer-group-count,state={empty|assigning|reconciling|stable|dead}
number of consumer groups based on state
consumer group rebalances sensor
kafka.server:type=group-coordinator-metrics,name=consumer-group-rebalance-rate
kafka.server:type=group-coordinator-metrics,name=consumer-group-rebalance-count
partition load sensor: __consumer_offsets partition load time
kafka.server:type=group-coordinator-metrics,name=partition-load-time-max
kafka.server:type=group-coordinator-metrics,name=partition-load-time-avg
thread idle ratio sensor: thread busy - idle ratio
kafka.server:type=group-coordinator-metrics,name=thread-idle-ratio-min
kafka.server:type=group-coordinator-metrics,name=thread-idle-ratio-avg
Group count by type
Group count by state
Rebalance Rate
Thread utilisation in percent

Broker Configurations

New properties in the broker configuration.

Name	Type	Default	Doc
group.coordinator.threads	int	1	The number of threads used to run the state machines.
group.consumer.session.timeout.ms	int	45s	The timeout to detect client failures when using the consumer group protocol.
group.consumer.min.session.timeout.ms	int	45s	The minimum session timeout.
group.consumer.max.session.timeout.ms	int	60s	The maximum session timeout.
group.consumer.heartbeat.interval.ms	int	5s	The heartbeat interval given to the members.
group.consumer.min.heartbeat.interval.ms	int	5s	The minimum heartbeat interval.
group.consumer.max.heartbeat.interval.ms	int	15s	The maximum heartbeat interval.
group.consumer.max.size	int	MaxValue	The maximum number of consumers that a single consumer group can accommodate.
group.consumer.assignors	listrange, uniform	org.apache.kafka.server.group.consumer.UniformAssignor, org.apache.kafka.server.group.consumer.RangeAssignor	The server side assignors as a list of full class names. The first one in the list is considered as the default assignor to be used in the case where the consumer does not specify an assignor.

...

When A heartbeats, the group coordinator transitions him it to its target epoch/assignment because it does not have any partitions to revoke. The group coordinator updates the member assignment and replies with the new epoch 1 and all the partitions.

...

When A heartbeats, the group coordinator instructs him it to revoke foo-2.

When A heartbeats again and acknowledges the revocation, the group coordinator transitions him it to epoch 2 and releases foo-2.

...

When B heartbeats, the group coordinator transitions him it to epoch 3 because B has no partitions to revoke. It persists the change and reply.

...

When A heartbeats, the group coordinator instructs him it to revoke foo-1.

When A heartbeats again and acknowledges the revocation, the group coordinator transitions him it to epoch 3 and releases foo-1.

...

C joins the group. The group coordinator adds himit, bumps the group epoch, create the member assignment, and computes the target assignment.

...

C heartbeats, the group coordinator transitions him it to epoch 22 but does not yet give him it any partitions because they are not revoked yet.

...

A heartbeats, the group coordinator instructs him it to revoke foo-2.

B heartbeats, the group coordinator instructs him it to revoke foo-5.

C heartbeats, no changes for himit.

A heartbeats and acknowledges the revocation, the group coordinator transitions him it to epoch 22, release foo-2, persists and reply.

...

C heartbeats, the group coordinator gives him it foo-2 which is now free but hold foo-5.

B heartbeats and acknowledges the revocation, the group coordinator transitions him it to epoch 22, releases foo-5, persists and reply.

...

C heartbeats, the group coordinator gives him it foo-2 and foo-5.

Member Failure

...

A fails to heartbeat. The group coordinator removes him it after the session timeout expires and bump the group epoch.

...

B leaves the group. The group coordinator removes him it and bumps the group epoch.

...

The group coordinator sees that C does not own any partitions any more, so it can transition to epoch 23 and transition to CompletingRebalance. The transition to epoch 23 is important here because the new epoch must be given to the member in the JoinGroup response. This is the new generation of the group for himit.

Group (epoch=23)

A (upgraded)
C (CompletingRebalance)

Target Assignment (epoch=23)

A - partitions=[foo-0, foo-1, foo-3]
C - partitions=[foo-2, foo-5, foo-4]

Member Assignment

A - epoch=22, partitions=[foo-0, foo-1], pending-partitions=[]
C - epoch=23, partitions=[foo-2, foo-5, foo-4], pending-partitions=[]

...

C sends the SyncGroup request and collects his its new assignment. All partitions are given because they are all free. C transitions to Stable.

...

Group (epoch=24)

A (upgraded)
B (upgraded)
C (Stable)

Target Assignment (epoch=24)

A - partitions=[foo-0, foo-1]
B - partitions=[foo-3, foo-4]
C - partitions=[foo-2, foo-5]

Member Assignment

A - epoch=24, partitions=[foo-0, foo-1], pending-partitions=[]
B - epoch=24, partitions=[foo-3, foo-4], pending-partitions=[]
C - epoch=24, partitions=[foo-2, foo-5], pending-partitions=[]

B heartbeats and gets his its assignment.

Compatibility, Deprecation, and Migration Plan

...

We considered storing the dynamic group configurations in the group coordinator in order to have the ability to tight their lifecycles to their group. We discarded this approach for two reasons: 1) This pattern does not fit very well in the IncrementalAlterConfig API as it would require to send updates about groups to the coordinator whereas all the other updates go to the controller; and 2) It seems preferable to decouple the life cycle of the dynamic configurations from the life cycle of the groups. Users may want to create configurations before the group is created and users may want to keep their configurations if the group is recreatedis recreated.

Client side generated Member ID

We considered letting the client generate its member ID (or UUID) instead of relying on the coordinator to generate one when the member joins the group. We ultimately rejected this because 1) it introduces extra dependencies on the client. For instance in C++, UUID are not natively supported so an extra library must be used; and 2) the client could not generate it correctly.

Future Work

Eventually, we aim at deprecating the current membership/rebalance API. In order to get to this point, we would need to first move all the use cases away from it.

...

The group membership protocol is also used outside of Apache Kafka. For instance, the Confluent Schema Registry uses it for leader election. It is not clear whether we really want to suppose such cases in the future. If we do, we could also define a new set of APIs for it. That would be much cleaner in the long run.

Metadata Transactions

The KIP proposes the rely on the atomicity of the batch to write the assignment to the __consumer_offsets partitions. This means that the assignment or, to be precise, the delta between two assignments can not be larger than 1MB where 1MB is the default batch size. In the future, we could imagine doing something similar to KIP-868 Metadata Transactions in the group coordinator. The solution outlined in KIP-868 does not work in our context because the __consumer_offsets is compacted. However, we could imagine a similar approach. We will tackle this in the future if needed.

Upgrade / Downgrade

The KIP proposes to rely on the IBP/MetadataVersion to decide whether a record or an API could be used or not. We have discussed the idea to use a dedicate feature flag instead of relying on metadata.version. That would allow decoupling the group coordinator from the quorum controller during upgrades. We also need to flush out how to handle downgrades. We will do this in a future KIP.

Space shortcuts

Child pages

Versions Compared

Old Version 82

New Version Current

Key

Status

Fencing

Consumer Group States

EMPTY

ConsumerGroupPrepareAssignment API

ConsumerGroupTargetAssignmentMetadataKey

ConsumerGroupTargetAssignmentMetadataValue

ConsumerGroupTargetAssignmentMemberKey

ConsumerGroupTargetAssignmentMemberValue

Current Member Assignment

Current Member Assignment

ConsumerGroupCurrentMemberAssignmentValue

Offsets

OffsetCommitValue

Broker Metrics

Broker Configurations

Member Failure

Compatibility, Deprecation, and Migration Plan

Client side generated Member ID

Future Work

Metadata Transactions

Upgrade / Downgrade

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 82

New Version Current

Key

Status

Fencing

Consumer Group States

EMPTY

ConsumerGroupPrepareAssignment API

ConsumerGroupTargetAssignmentMetadataKey

ConsumerGroupTargetAssignmentMetadataValue

ConsumerGroupTargetAssignmentMemberKey

ConsumerGroupTargetAssignmentMemberValue

Current Member Assignment

Current Member Assignment

ConsumerGroupCurrentMemberAssignmentValue

Offsets

OffsetCommitValue

Broker Metrics

Broker Configurations

Member Failure

Compatibility, Deprecation, and Migration Plan

Client side generated Member ID

Future Work

Metadata Transactions

Upgrade / Downgrade