dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r771248687
##########
File path: clients/src/main/resources/common/message/LeaveGroupResponse.json
##########
@@ -24,7 +24,9 @@
// Starting in version 3, we will make leave group request into batch mode
and add group.instance.id.
//
// Version 4 is the first flexible version.
- "validVersions": "0-4",
+ //
+ // Version 5 adds the Reason field (KIP-800).
Review comment:
nit: Should we rather say `Version 5 is the same as version 4.` here?
##########
File path:
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1837,7 +1837,12 @@ private DescribeGroupsResponse
createDescribeGroupResponse() {
}
private LeaveGroupRequest createLeaveGroupRequest(short version) {
- return new LeaveGroupRequest.Builder("group1", singletonList(new
MemberIdentity().setMemberId("consumer1")))
+ MemberIdentity member = new MemberIdentity()
+ .setMemberId("consumer1");
Review comment:
nit: I think that we can keep this one on the previous line.
##########
File path:
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
##########
@@ -1837,7 +1837,12 @@ private DescribeGroupsResponse
createDescribeGroupResponse() {
}
private LeaveGroupRequest createLeaveGroupRequest(short version) {
- return new LeaveGroupRequest.Builder("group1", singletonList(new
MemberIdentity().setMemberId("consumer1")))
+ MemberIdentity member = new MemberIdentity()
+ .setMemberId("consumer1");
+ if (version >= 5) {
+ member.setMemberId("reason: test");
Review comment:
This is not correct.
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -617,9 +617,10 @@ class GroupCoordinator(val brokerId: Int,
leavingMembers: List[MemberIdentity],
responseCallback: LeaveGroupResult => Unit): Unit = {
- def removeCurrentMemberFromGroup(group: GroupMetadata, memberId: String):
Unit = {
+ def removeCurrentMemberFromGroup(group: GroupMetadata, memberId: String,
reason: Option[String]): Unit = {
val member = group.get(memberId)
- removeMemberAndUpdateGroup(group, member, s"Removing member $memberId on
LeaveGroup")
+ val leaveReason = reason.getOrElse("unknown reason")
+ removeMemberAndUpdateGroup(group, member, s"Removing member $memberId on
LeaveGroup due to: $leaveReason")
removeHeartbeatForLeavingMember(group, member.memberId)
info(s"Member $member has left group $groupId through explicit
`LeaveGroup` request")
Review comment:
I am also tempted to add the reason here. What do you think?
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3907,6 +3909,42 @@ public void testRemoveMembersFromGroup() throws
Exception {
}
}
+ @Test
+ public void testRemoveMembersFromGroupReason() throws Exception {
+ final Cluster cluster = mockCluster(3, 0);
+ final Time time = new MockTime();
+
+ try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time,
cluster)) {
+
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+
+
env.kafkaClient().prepareResponse(prepareFindCoordinatorResponse(Errors.NONE,
env.cluster().controller()));
+ env.kafkaClient().prepareResponse(body -> {
+ if (!(body instanceof LeaveGroupRequest)) {
+ return false;
+ }
+ LeaveGroupRequestData leaveGroupRequest = ((LeaveGroupRequest)
body).data();
+
+ return leaveGroupRequest.members().stream().allMatch(member ->
member.reason().equals("testing remove members reason"));
+ }, new LeaveGroupResponse(new
LeaveGroupResponseData().setErrorCode(Errors.NONE.code()).setMembers(
+ Arrays.asList(
+ new
MemberResponse().setGroupInstanceId("instance-1"),
+ new
MemberResponse().setGroupInstanceId("instance-2")
+ ))
Review comment:
nit: Indentation of those lines seems to be off here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
