Status

Current state: Under Discussion

Discussion thread: here

JIRA:

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

Motivation

KIP-848: The Next Generation of the Consumer Rebalance Protocol introduces new status in consumer group like group epoch, target assignment epoch, member epoch, and target assignment. Adding this information to kafka-consumer-groups.sh and kafka-share-groups.sh commands can help administrator to have a detailed view of the state of the groups.

Public Interfaces

Client API changes

Admin

ConsumerGroupDescription

Add new fields groupEpoch and targetAssignmentEpoch. The ConsumerGroupDescription can be used for classic and consumer group. For classic group, there is no epoch information, so using Optional for new fields.

public class ConsumerGroupDescription {
    private final String groupId;
    private final boolean isSimpleConsumerGroup;
    private final Collection<MemberDescription> members;
    private final String partitionAssignor;
    private final GroupType type;
    private final ConsumerGroupState state;
    private final Node coordinator;
    private final Set<AclOperation> authorizedOperations;
	private final Optional<Integer> groupEpoch;
    private final Optional<Integer> targetAssignmentEpoch;

	// ...

	/**
	 * The epoch of the consumer group.
	 */
	public Optional<Integer> groupEpoch() {
		return groupEpoch;
	}

	/**
	 * The epoch of the target assignment.
	 */
	public Optional<Integer> targetAssignmentEpoch() {
		return targetAssignmentEpoch;
	}
}

ShareGroupDescription

Add new fields groupEpoch and targetAssignmentEpoch.

public class ShareGroupDescription {
    private final String groupId;
    private final Collection<MemberDescription> members;
    private final ShareGroupState state;
    private final Node coordinator;
    private final Set<AclOperation> authorizedOperations;
	private final int groupEpoch;
    private final int targetAssignmentEpoch;

 	// ...

	/**
	 * The epoch of the share group.
	 */
	public int groupEpoch() {
		return groupEpoch;
	}

	/**
	 * The epoch of the target assignment.
	 */
	public int targetAssignmentEpoch() {
		return targetAssignmentEpoch;
	}
}

MemberDescription

Add a new field memberEpoch. For classic group member, there is no member epoch, so using Optional for the new field.

public class MemberDescription {
    private final String memberId;
    private final Optional<String> groupInstanceId;
    private final String clientId;
    private final String host;
    private final MemberAssignment assignment;
    private final Optional<MemberAssignment> targetAssignment;
    private final Optional<Integer> memberEpoch;

	// ...

	/**
	 * The epoch of the group member.
	 */
	public Optional<Integer> memberEpoch() {
		return memberEpoch;
	}
}

Proposed Changes

kafka-consumer-groups.sh

--describe --verbose

Show the group level information.

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --verbose
GROUP		COORDINATOR (ID)          ASSIGNMENT-STRATEGY  	STATE           	#MEMBERS 	GROUP-EPOCH		TARGET-ASSIGNMENT-EPOCH
my-group	localhost:61316  (0)      uniform      		 	Stable			 	1			1				1

--describe --members --verbose

Show the member level information. Add GROUP-EPOCH, TARGET-ASSIGNMENT-EPOCH, MEMBER-EPOCH, and TARGET-ASSIGNMENT information.

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
GROUP		CONSUMER-ID            HOST            CLIENT-ID			#PARTITIONS		ASSIGNMENT		TARGET-ASSIGNMENT		GROUP-EPOCH 	TARGET-ASSIGNMENT-EPOCH		MEMBER-EPOCH
my-group	T4tbGKsvT7CtsxVBH5J2QQ /127.0.0.1      consumer-my-group-1	1				(0)				(0)						1				1							1

--describe --offsets --verbose

Show leader epoch which was introduced in KIP-320.

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --offsets --verbose
GROUP       TOPIC    	PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG            CONSUMER-ID                                   	HOST            CLIENT-ID									LEADER-EPOCH
my-group	my-topic 	0          10              10              10             my-group-3056e339-59e1-43d8-abe3-8a9e10c53937 	/127.0.0.1      consumer-test.group.CLASSIC.--offsets-2		1

kafka-share-group.sh

--describe --verbose

Show the group level information.

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state
GROUP		COORDINATOR (ID)          STATE			#MEMBERS 	GROUP-EPOCH		TARGET-ASSIGNMENT-EPOCH
my-group	localhost:61316  (0)      Stable			1			1				1

--describe --members --verbose

Show the member level information. Add GROUP-EPOCH, TARGET-ASSIGNMENT-EPOCH, MEMBER-EPOCH, and TARGET-ASSIGNMENT information.

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members
GROUP		CONSUMER-ID            HOST            CLIENT-ID  				ASSIGNMENT		TARGET-ASSIGNMENT	GROUP-EPOCH 	TARGET-ASSIGNMENT-EPOCH  	MEMBER-EPOCH
my-group	T4tbGKsvT7CtsxVBH5J2QQ /127.0.0.1      consumer-my-group-1		my_topic:0		my_topic:0			1				1							1

Compatibility, Deprecation, and Migration Plan

This proposal adds new optional fields to Admin client and kafka-consumer-groups.sh, so classic consumer groups will be unaffected.

Test Plan

The feature will be thoroughly tested with unit and integration tests.

Rejected Alternatives

N/A