frankvicky commented on code in PR #17614:
URL: https://github.com/apache/kafka/pull/17614#discussion_r1972859960
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##########
@@ -427,7 +428,7 @@ public void onGroupAssignmentUpdated(Set<TopicPartition>
partitions) {
// call close methods if internal objects are already constructed;
this is to prevent resource leak. see KAFKA-2121
// we do not need to call `close` at all when `log` is null, which
means no internal objects were initialized.
if (this.log != null) {
- close(Duration.ZERO, true);
+ close(Duration.ZERO,
CloseOptions.GroupMembershipOperation.DEFAULT, true);
Review Comment:
IIUC, `AsyncKafkaConsumer` doesn't have a single RPC specifically
responsible for leaving the group. This functionality is piggybacked on the
`ConsumerGroupHeartbeat` by assigning memberEpoch as `-1` for leaving the
group. [0]
But you're pointing out a potential issue - perhaps we should consider using
`LEAVE_GROUP` instead of `DEFAULT`?
In this case, consumerGroupHeartbeat will send if heartbeatRequestManager is
created, so it may be better to send the request with a leaving epoch. The
server will throw `UnknownMemberIdException` if the member doesn't exit.[1] I
think it's ok for this case.
[0]
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=217387038#KIP848:TheNextGenerationoftheConsumerRebalanceProtocol-Joining&Leaving
[1]
https://github.com/apache/kafka/blob/8bbca913efe260ba59c824801466f73584c46f8f/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/consumer/ConsumerGroup.java#L257
--
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
