You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/12/07 05:01:19 UTC
[GitHub] [kafka] jeffkbkim opened a new pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
jeffkbkim opened a new pull request #11571:
URL: https://github.com/apache/kafka/pull/11571
As part of [KIP-800](https://cwiki.apache.org/confluence/display/KAFKA/KIP-800%3A+Add+reason+to+JoinGroupRequest+and+LeaveGroupRequest), this patch aims to add a reason to why a consumer is leaving the group and log it in the broker.
for JoinGroupRequest: https://github.com/apache/kafka/pull/11566
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r769741518
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3731,6 +3731,11 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
} else {
members = options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
}
+
+ if (options.reason() != null) {
+ members.forEach(member -> member.setReason(options.reason()));
Review comment:
I think the KIP also specifies a default reason in this case, doesn't it?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -647,7 +647,7 @@ class GroupCoordinator(val brokerId: Int,
if (memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID) {
groupInstanceId.flatMap(group.currentStaticMemberId) match {
case Some(currentMemberId) =>
- removeCurrentMemberFromGroup(group, currentMemberId)
+ removeCurrentMemberFromGroup(group, currentMemberId, leavingMember.reason())
Review comment:
nit: Parenthesis could be omitted.
##########
File path: clients/src/main/resources/common/message/LeaveGroupRequest.json
##########
@@ -36,7 +38,10 @@
"about": "The member ID to remove from the group." },
{ "name": "GroupInstanceId", "type": "string",
"versions": "3+", "nullableVersions": "3+", "default": "null",
- "about": "The group instance ID to remove from the group." }
+ "about": "The group instance ID to remove from the group." },
Review comment:
Could we extend tests in `RequestResponseTest` to cover this new field?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -617,9 +617,9 @@ 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, leaveReason: String): Unit = {
Review comment:
Should we use `Option[String]` for the `leaveReason` and handle the case where it is not provided?
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -665,7 +665,7 @@ class GroupCoordinator(val brokerId: Int,
groupInstanceId,
operation = "leave-group"
).getOrElse {
- removeCurrentMemberFromGroup(group, memberId)
+ removeCurrentMemberFromGroup(group, memberId, leavingMember.reason())
Review comment:
ditto.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-989475604
@dajac the errors come from https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaApis.scala#L1833:
```
new LeaveGroupResponse(
memberResponses.asJava,
leaveGroupResult.topLevelError,
requestThrottleMs,
leaveGroupRequest.version)
```
where we pass in the leave group request version to create the leave group response. is this behavior expected? if so, should we update the valid versions in LeaveGroupResponse.json from 0-4 to 0-5? (5 is the new version in LeaveGroupRequest from KIP-800)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-1007608090
@dajac please take another pass when you get the chance
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r772792590
##########
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:
thanks for the catch!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r782071080
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3907,6 +3909,44 @@ 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("member was removed by an admin: 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")
+ ))
+ ));
+
+ Collection<MemberToRemove> membersToRemove = Arrays.asList(new MemberToRemove("instance-1"), new MemberToRemove("instance-2"));
+
+ RemoveMembersFromConsumerGroupOptions options = new RemoveMembersFromConsumerGroupOptions(membersToRemove);
+ options.reason("testing remove members reason");
Review comment:
Should we also add a test that does not specify the reason to test the default case?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r770867250
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java
##########
@@ -45,10 +46,21 @@ public RemoveMembersFromConsumerGroupOptions() {
this.members = Collections.emptySet();
}
+ /**
+ * Sets an optional reason.
+ */
+ public void reason(final String reason) {
+ this.reason = reason;
Review comment:
We have to follow the KIP on the naming here. We usually don't use "set" 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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-1007608090
@dajac please take another pass when you get the chance
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim edited a comment on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim edited a comment on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-988460630
tests calling `adminClient.removeMembersFromConsumerGroup` are failing:
```
ERROR [KafkaApi-2] Unexpected error handling request RequestHeader(apiKey=LEAVE_GROUP, apiVersion=5, clientId=adminclient-1, correlationId=30) -- LeaveGroupRequestData(groupId='test_group_id', memberId='', members=[MemberIdentity(memberId='', groupInstanceId='invalid-instance-id', reason='null')]) with context RequestContext(header=RequestHeader(apiKey=LEAVE_GROUP, apiVersion=5, clientId=adminclient-1, correlationId=30), connectionId='127.0.0.1:58238-127.0.0.1:58241-0', clientAddress=/127.0.0.1, principal=User:ANONYMOUS, listenerName=ListenerName(PLAINTEXT), securityProtocol=PLAINTEXT, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=unknown), fromPrivilegedListener=true, principalSerde=Optional[org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder@5d777a17]) (kafka.server.KafkaApis:76)
org.apache.kafka.common.errors.UnsupportedVersionException: Can't size version 5 of MemberResponse
```
i'm not sure i understand the exception. (key=LEAVE_GROUP,api version=5) requests are valid.
the error is from `LeaveGroupResponseData.addSize` - why is the response validating the request api version?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r783200432
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3907,6 +3910,79 @@ 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(LEAVE_GROUP_REASON + ": 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")
+ ))
+ ));
+
+ Collection<MemberToRemove> membersToRemove = Arrays.asList(new MemberToRemove("instance-1"), new MemberToRemove("instance-2"));
+
+ RemoveMembersFromConsumerGroupOptions options = new RemoveMembersFromConsumerGroupOptions(membersToRemove);
+ options.reason("testing remove members reason");
+
+ final RemoveMembersFromConsumerGroupResult result = env.adminClient().removeMembersFromConsumerGroup(
+ GROUP_ID, options);
+
+ assertNull(result.all().get());
+ }
+ }
+
+ @Test
+ public void testRemoveMembersFromGroupDefaultReason() throws Exception {
Review comment:
you're totally right, will update
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-1007872215
@jeffkbkim , this tests `testRemoveMembersFromGroupReason()` is still failed. Please help fix it. Thanks.
https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11571/11/#showFailuresLink
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-988460630
tests calling `adminClient.removeMembersFromConsumerGroup` are failing:
```
ERROR [KafkaApi-2] Unexpected error handling request RequestHeader(apiKey=LEAVE_GROUP, apiVersion=5, clientId=adminclient-1, correlationId=30) -- LeaveGroupRequestData(groupId='test_group_id', memberId='', members=[MemberIdentity(memberId='', groupInstanceId='invalid-instance-id', reason='null')]) with context RequestContext(header=RequestHeader(apiKey=LEAVE_GROUP, apiVersion=5, clientId=adminclient-1, correlationId=30), connectionId='127.0.0.1:58238-127.0.0.1:58241-0', clientAddress=/127.0.0.1, principal=User:ANONYMOUS, listenerName=ListenerName(PLAINTEXT), securityProtocol=PLAINTEXT, clientInformation=ClientInformation(softwareName=apache-kafka-java, softwareVersion=unknown), fromPrivilegedListener=true, principalSerde=Optional[org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder@5d777a17]) (kafka.server.KafkaApis:76)
org.apache.kafka.common.errors.UnsupportedVersionException: Can't size version 5 of MemberResponse
```
i'm not sure i understand the exception. LEAVE_GROUP/API_VERSION=5 requests are valid.
the error is from `LeaveGroupResponseData.addSize` - why is the response validating the request api version?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-991664350
@jeffkbkim That's right. We have to bump the version of both the request and the response.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #11571: KAFKA-13496: Add reason to LeaveGroupRequest (KIP-800)
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #11571:
URL: https://github.com/apache/kafka/pull/11571
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-991664350
@jeffkbkim That's right. We have to bump the version of both the request and the response.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r770997666
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java
##########
@@ -45,10 +46,21 @@ public RemoveMembersFromConsumerGroupOptions() {
this.members = Collections.emptySet();
}
+ /**
+ * Sets an optional reason.
+ */
+ public void reason(final String reason) {
+ this.reason = reason;
Review comment:
yeah, i was on the fence for this. reverting
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim edited a comment on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
jeffkbkim edited a comment on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-988460630
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r771376907
##########
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")
Review comment:
I think that we should use the same pattern for both join reasons and leave reason. See https://github.com/apache/kafka/pull/11566#discussion_r771371068.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r770255842
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3731,6 +3731,11 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
} else {
members = options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
}
+
+ String baseReason = "member was removed by an admin";
+ String reason = options.reason() == null ? baseReason : baseReason + ": " + options.reason();
Review comment:
I think the `baseReason` can be declared as `static` variable.
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java
##########
@@ -45,10 +46,21 @@ public RemoveMembersFromConsumerGroupOptions() {
this.members = Collections.emptySet();
}
+ /**
+ * Sets an optional reason.
+ */
+ public void reason(final String reason) {
+ this.reason = reason;
Review comment:
We have a `reason` setter and getter in the class. I think we should name them differently for other developer better distinguishing between them, instead of overloading them with different purpose. how about `setReason` and `reason`?
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3731,6 +3731,11 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
} else {
members = options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
}
+
+ String baseReason = "member was removed by an admin";
+ String reason = options.reason() == null ? baseReason : baseReason + ": " + options.reason();
Review comment:
And maybe rename to `baseLeaveGroupReason` or other names for much specific.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
showuon commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r771167832
##########
File path: clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java
##########
@@ -45,10 +46,21 @@ public RemoveMembersFromConsumerGroupOptions() {
this.members = Collections.emptySet();
}
+ /**
+ * Sets an optional reason.
+ */
+ public void reason(final String reason) {
+ this.reason = reason;
Review comment:
OK, if it's following the KIP, then I'm fine with it. Thanks.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#issuecomment-1007872215
@jeffkbkim , this tests `testRemoveMembersFromGroupReason()` is still failed. Please help fix it. Thanks.
https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-11571/11/#showFailuresLink
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #11571: KAFKA-13496: add reason to LeaveGroupRequest
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #11571:
URL: https://github.com/apache/kafka/pull/11571#discussion_r783185731
##########
File path: clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -3907,6 +3910,79 @@ 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(LEAVE_GROUP_REASON + ": 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")
+ ))
+ ));
+
+ Collection<MemberToRemove> membersToRemove = Arrays.asList(new MemberToRemove("instance-1"), new MemberToRemove("instance-2"));
+
+ RemoveMembersFromConsumerGroupOptions options = new RemoveMembersFromConsumerGroupOptions(membersToRemove);
+ options.reason("testing remove members reason");
+
+ final RemoveMembersFromConsumerGroupResult result = env.adminClient().removeMembersFromConsumerGroup(
+ GROUP_ID, options);
+
+ assertNull(result.all().get());
+ }
+ }
+
+ @Test
+ public void testRemoveMembersFromGroupDefaultReason() throws Exception {
Review comment:
The code duplication is a bit annoying here, don't you think? I wonder if we could define an helper method, say `testRemoveMembersFromGroup(String reason, String expectedReason)`. Then we could call it from within the two test cases:
* `testRemoveMembersFromGroup("testing remove members reason", LEAVE_GROUP_REASON + ": testing remove members reason")`
* `testRemoveMembersFromGroup(null, LEAVE_GROUP_REASON)` - passing `null` is somewhat equivalent to not setting the reason at all. Otherwise, we could pass the `RemoveMembersFromConsumerGroupOptions` to the helper method.
What do you think?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org