jeffkbkim commented on a change in pull request #11971:
URL: https://github.com/apache/kafka/pull/11971#discussion_r839055468
##########
File path:
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
##########
@@ -2819,6 +2851,39 @@ public void
testEnforceRebalanceTriggersRebalanceOnNextPoll() {
assertEquals(countingRebalanceListener.revokedCount, 1);
}
+ @Test
+ public void testEnforceRebalanceReason() {
+ Time time = new MockTime(1L);
+ ConsumerMetadata metadata = createMetadata(subscription);
+ MockClient client = new MockClient(time, metadata);
+ KafkaConsumer<String, String> consumer = newConsumer(time, client,
subscription, metadata, assignor, true, groupInstanceId);
+ MockRebalanceListener countingRebalanceListener = new
MockRebalanceListener();
+ initMetadata(client, Utils.mkMap(Utils.mkEntry(topic, 1),
Utils.mkEntry(topic2, 1), Utils.mkEntry(topic3, 1)));
+
+ consumer.subscribe(Arrays.asList(topic, topic2),
countingRebalanceListener);
+ Node node = metadata.fetch().nodes().get(0);
+ prepareRebalance(client, node, assignor, Arrays.asList(tp0, t2p0),
null);
+
+ // a first rebalance to get the assignment, we need two poll calls
since we need two round trips to finish join / sync-group
+ consumer.poll(Duration.ZERO);
+ consumer.poll(Duration.ZERO);
Review comment:
the test passes without the second poll. the first poll finishes the sync
```
INFO Successfully synced group in generation
```
before the second poll is triggered.
The second poll notifies the assignor and gets committed offsets which i
don't think is necessary in this test
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3723,27 +3723,34 @@ private Integer nodeFor(ConfigResource resource) {
List<MemberIdentity> membersToRemove = new ArrayList<>();
for (final MemberDescription member : members) {
+ MemberIdentity memberIdentity = new MemberIdentity()
+ .setReason(reason);
+
if (member.groupInstanceId().isPresent()) {
- membersToRemove.add(new
MemberIdentity().setGroupInstanceId(member.groupInstanceId().get()));
+
memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
} else {
- membersToRemove.add(new
MemberIdentity().setMemberId(member.consumerId()));
+ memberIdentity.setMemberId(member.consumerId());
}
+
+ membersToRemove.add(memberIdentity);
}
return membersToRemove;
}
@Override
public RemoveMembersFromConsumerGroupResult
removeMembersFromConsumerGroup(String groupId,
RemoveMembersFromConsumerGroupOptions options) {
+ String reason = options.reason() == null || options.reason().isEmpty()
?
+ DEFAULT_LEAVE_GROUP_REASON : options.reason();
+
List<MemberIdentity> members;
if (options.removeAll()) {
- members = getMembersFromGroup(groupId);
+ members = getMembersFromGroup(groupId, reason);
} else {
- members =
options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
+ members = options.members().stream()
+ .map(m -> m.toMemberIdentity().setReason(reason))
+ .collect(Collectors.toList());
Review comment:
should this have been done as part of KIP-800?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3723,27 +3723,34 @@ private Integer nodeFor(ConfigResource resource) {
List<MemberIdentity> membersToRemove = new ArrayList<>();
for (final MemberDescription member : members) {
+ MemberIdentity memberIdentity = new MemberIdentity()
+ .setReason(reason);
Review comment:
nit: does `.setReason()` have to be in its own line?
--
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
