Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

Table of Contents

Status

Current state: Under Discussion Accepted

Discussion thread: here, vote thread

...

Add new fields memberEpoch and protocolupgraded. For classic group member, there is no member epoch, so using Optional for the new field. For upgraded field, the value is like following:

  • Classic Group -> Optional.empty
  • Consumer Group -> Optional.empty if unknown
  • Consumer Group -> Optional.of(false) if classic
  • Consumer Group -> Optional.of(true) if consumer

Deprecated old constructors.

Code Block
languagejava
titleMemberDescription
linenumberstrue
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;
    private final StringOptional<Boolean> protocolupgraded;

	// ...

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

    /**
public MemberDescription(String memberId,
                        * The protocol of the groupOptional<String> member.groupInstanceId,
     */
     public String protocol() {
        return protocol;
    }
}

RPCs

ConsumerGroupDescribeRequest

Bump validVersions to "0-1".

Code Block
titleConsumerGroupDescribeRequest
{
  "apiKey": 69   String clientId,
  "type": "request",
  "listeners": ["broker"],
  "name": "ConsumerGroupDescribeRequest",
  "validVersions": "0-1",  <-- bump to 0-1 
  "flexibleVersions": "0+",
  "fields": [
    { "name": "GroupIds", "type": "[]string", "versions": "0+", "entityType": "groupId",
String host,
            "about": "The ids of the groups to describe" },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+",
 MemberAssignment assignment,
                       "about": "Whether to include authorized operations." }
  ]
}

ConsumerGroupDescribeResponse

Bump validVersions to "0-1" and add a new field "Protocol" to "Members".

Code Block
titleConsumerGroupDescribeResponse
{
  "apiKey": 69,
  "type": "response",
  "name": "ConsumerGroupDescribeResponse",
  "validVersions": "0-1",  <-- bump to 0-1
  "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+",      Optional<MemberAssignment> targetAssignment,
                             Optional<Integer> memberEpoch,
                             Optional<Boolean> upgraded
    ) {
        this.memberId = memberId == null ? "" : memberId;
        this.groupInstanceId = groupInstanceId;
      "about": "The durationthis.clientId in= millisecondsclientId for== whichnull the? request"" was: throttledclientId;
 due to a quota violation, or zero ifthis.host the= requesthost did== notnull violate? any quota." },
"" : host;
     { "name": "Groups", "type": "[]DescribedGroup", "versions": "0+",   this.assignment = assignment == null ?
      "about": "Each described group.",
   new   "fields": [MemberAssignment(Collections.emptySet()) : assignment;
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
this.targetAssignment = targetAssignment;
        this.memberEpoch = memberEpoch;
   "about": "The describe error, or 0this.upgraded if= thereupgraded;
 was no error." },

    /**
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",* @deprecated Since 4.0. Use {@link #MemberDescription(String, Optional, String, String, MemberAssignment, Optional, Optional, Optional)}.
     */
    @Deprecated
 "about": "The top-level error message, or null if there was no error." },
public MemberDescription(String memberId,
                      { "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId" Optional<String> groupInstanceId,
          "about": "The group ID string." },
              { "name": "GroupState", "type": "string", "versions": "0+" String clientId,
          "about": "The group state string, or the empty string." },
                  { "name": "GroupEpoch", "type": "int32", "versions": "0+",
String host,
               "about": "The group epoch." },
             { "name": "AssignmentEpoch", "type": "int32", "versions": "0+",
MemberAssignment assignment,
              "about": "The assignment epoch." },
        { "name": "AssignorName", "type": "string", "versions": "0+",
    Optional<MemberAssignment> targetAssignment
    ) {
        this(
       "about": "The selected assignor." },
       memberId,
      { "name": "Members", "type": "[]Member", "versions": "0+" groupInstanceId,
          "about": "The members."clientId,
          "fields": [  host,
            { "name": "MemberId", "type": "string", "versions": "0+",
assignment,
            targetAssignment,
         "about": "The member ID." }Optional.empty(),
            Optional.empty()
 { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",    );
    }

    /**
     * @deprecated Since 4.0.  "about": "The member instance ID." },
   Use {@link #MemberDescription(String, Optional, String, String, MemberAssignment, Optional, Optional, Optional)}.
     */
    { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null"@Deprecated
    public MemberDescription(
        String memberId,
        Optional<String> groupInstanceId,
     "about": "The member rack ID." }String clientId,
        String host,
   { "name": "MemberEpoch", "type": "int32", "versions": "0+",     MemberAssignment assignment
    ) {
        this(
      "about": "The current member epoch." }      memberId,
            { "name": "ClientId", "type": "string", "versions": "0+",
groupInstanceId,
            clientId,
        "about": "The client ID." }host,
            { "name": "ClientHost", "type": "string", "versions": "0+",
assignment,
            Optional.empty()
      "about": "The client host." }, );
    }

    /**
     * @deprecated Since 4.0. Use { "name": "SubscribedTopicNames", "type": "[]string", "versions": "0+", "entityType": "topicName",@link #MemberDescription(String, Optional, String, String, MemberAssignment, Optional, Optional, Optional)}.
      */
    @Deprecated
    public  "about": "The subscribed topic names." },
MemberDescription(String memberId,
                    { "name": "SubscribedTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null" String clientId,
               "about": "the subscribed topic regex otherwise or null of not provided." },
    String host,
       { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The currentMemberAssignment assignment.") },{
        this(memberId,    { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",Optional.empty(), clientId, host, assignment);
    }

	// ...

	/**
	 * The epoch of the group member.
	  "about": "The target assignment." },*/
	public Optional<Integer> memberEpoch() {
		return memberEpoch;
	}

    /**
     * The flag { "name": "Protocol", "type": "string", "versions": "1+", "ignorable": true,indicating whether a member is classic.
     */
    public Optional<Boolean> upgraded() {
        "about": "The member protocol." }  <-- new field
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructsreturn upgraded;
    }
}

RPCs

ConsumerGroupDescribeRequest

Bump validVersions to "0-1".

Code Block
titleConsumerGroupDescribeRequest
{
  "apiKey": 69,
  "type": "request",
  "listeners": ["broker"],
  "name": "ConsumerGroupDescribeRequest",
  "validVersions": "0-1",  <-- bump to 0-1 
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicPartitionsGroupIds", "versionstype": "0+", "fields": [
      { "name": "TopicId"[]string", "typeversions": "uuid0+", "versionsentityType": "0+groupId",
        "about": "The topic ID. ids of the groups to describe" },
      { "name": "TopicNameIncludeAuthorizedOperations", "type": "stringbool", "versions": "0+", "entityType": "topicName",
        "about": "The topic nameWhether to include authorized operations." },
  ]
}


ConsumerGroupDescribeResponse

Bump validVersions to "0-1" and add a new field "MemberType" to "Members". The "MemberType" field is int8. It's -1 as default for unknown, 0 for classic member, and 1 for consumer member.

Code Block
titleConsumerGroupDescribeResponse
{
  "apiKey": 69,
  "type": "response",
     { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partitions." }
    ]},
    { "name": "AssignmentConsumerGroupDescribeResponse",
  "versionsvalidVersions": "0+-1", "fields": [
      { "name <-- bump to 0-1
  "flexibleVersions": "TopicPartitions0+", "type": "[]TopicPartitions", "versions": "0+",
  // Supported errors:
  // - GROUP_AUTHORIZATION_FAILED (version 0+)
  "about": "The assigned topic-partitions to the member." }
    ]}
  ]
}

Proposed Changes

kafka-consumer-groups.sh

--describe --state --verbose

Show the group level information. Add GROUP-EPOCH and TARGET-ASSIGNMENT-EPOCH. If the value is empty, showing "-" instead.

// - 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": "AssignorName", "type": "string", "versions": "0+",
          "about": "The selected assignor." },
        { "name": "Members", "type": "[]Member", "versions": "0+",
          "about": "The members.",
          "fields": [
            { "name": "MemberId", "type": "string", "versions": "0+",
              "about": "The member ID." },
            { "name": "InstanceId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member instance ID." },
            { "name": "RackId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "The member rack ID." },
            { "name": "MemberEpoch", "type": "int32", "versions": "0+",
              "about": "The current member epoch." },
            { "name": "ClientId", "type": "string", "versions": "0+",
              "about": "The client ID." },
            { "name": "ClientHost", "type": "string", "versions": "0+",
              "about": "The client host." },
            { "name": "SubscribedTopicNames", "type": "[]string", "versions": "0+", "entityType": "topicName",
              "about": "The subscribed topic names." },
            { "name": "SubscribedTopicRegex", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
              "about": "the subscribed topic regex otherwise or null of not provided." },
            { "name": "Assignment", "type": "Assignment", "versions": "0+",
              "about": "The current assignment." },
            { "name": "TargetAssignment", "type": "Assignment", "versions": "0+",
              "about": "The target assignment." },
            { "name": "MemberType", "type": "int8", "versions": "1+", "default": "-1", "ignorable": true,
              "about": "-1 for unknown. 0 for classic member. +1 for consumer member." }  <-- new field
          ]},
        { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
          "about": "32-bit bitfield to represent authorized operations for this group." }
      ]
    }
  ],
  "commonStructs": [
    { "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": "TopicPartitions", "type": "[]TopicPartitions", "versions": "0+",
        "about": "The assigned topic-partitions to the member." }
    ]}
  ]
}

Proposed Changes

kafka-consumer-groups.sh

--describe --state --verbose

Show the group level information. Add GROUP-EPOCH and TARGET-ASSIGNMENT-EPOCH. If the value is empty, showing "-" instead.

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

--describe --members --verbose

Show the member level information. Add CURRENT-EPOCH, TARGET-EPOCH, TARGET-ASSIGNMENT, and PROTOCOL information. Change ASSIGNMENT to CURRENT-ASSIGNMENT, TARGET-EPOCH, TARGET-ASSIGNMENT, and UPGRADED information. The CURRENT-EPOCH means member epoch. Change ASSIGNMENT to CURRENT-ASSIGNMENT. For both classic and consumer groups, the ASSIGNMENT value format will change from "(0,1), (0,1)" to "my_topic:0,1;new_topic:0,1". The original format only shows sorted partition numbers for each topic. The new format adds the topic name. The UPGRADED is showed when a group is in migration.

Code Block
$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
GROUP			CONSUMER-ID				HOST			CLIENT-ID			#PARTITIONS		CURRENT-EPOCH	CURRENT-ASSIGNMENT			TARGET-EPOCH	TARGET-ASSIGNMENT			PROTOCOLUPGRADED
consumermy-group		T4tbGKsvT7CtsxVBH5J2QQ	/127.0.0.1		consumer-group-1	4member		3				13				my_topic:0,1;new_topic:0,1	13				my_topic:0,1;new_topic:0,1	consumertrue
classicmy-group	984f918d0921caae	6mfoIHq4n3BT7n1HCdqihb	/127.0.0.1		classic-group-1member		1				42				-	1					my_topic:0,1;new_topic:0,1	1	3				my_topic:0,1;new_topic:0,1					classicfalse

Following table lists protocol UPGRADED meaning.

ProtocolUPGRADEDMeaning
"classictrue"

A consumer member in classic group.A member with non-null classicMemberMetadata in a consumer group.

"consumerfalse"A classic member with null classicMemberMetadata in consumer group."share"A member in share a consumer group.

--describe --offsets --verbose / --describe --verbose

...

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

kafka-share-groups.sh

Add a new "–verbose" option to ShareGroupCommandOptions.

...

Show the group level information. Add GROUP-EPOCH and TARGET- ASSIGNMENT-EPOCH.

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

--describe --members --verbose

Show the member level information. Add CURRENT-EPOCH, TARGET-EPOCH, and TARGET-ASSIGNMENT information. Change ASSIGNMENT to CURRENT-ASSIGNMENTMEMBER-EPOCH and ASSIGNMENT information. The ASSIGNMENT value format will change from "my_topic:0,my_topic:1,new_topic:0,new_topic:1" to "my_topic:0,1;new_topic:0,1". The original format shows topic name for each partition. The new format only shows topic name one time for all partitions under the same topic.

Code Block
$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose
GROUP		CONSUMER-ID				HOST			CLIENT-ID				CURRENTMEMBER-EPOCH	CURRENT-ASSIGNMENT			TARGET-EPOCH	TARGET-ASSIGNMENT
my-group	T4tbGKsvT7CtsxVBH5J2QQ	/127.0.0.1		consumer-my-group-1		1				my_topic:0,1;new_topic:0,1	1				my_topic:0,1;new_topic:0,1

...