frankvicky commented on code in PR #17614:
URL: https://github.com/apache/kafka/pull/17614#discussion_r1972757524
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##########
@@ -1151,31 +1153,28 @@ protected void handlePollTimeoutExpiry() {
"either by increasing max.poll.interval.ms or by reducing the
maximum size of batches " +
"returned in poll() with max.poll.records.");
- maybeLeaveGroup("consumer poll timeout has expired.");
+ maybeLeaveGroup(DEFAULT, "consumer poll timeout has expired.");
}
/**
- * Sends LeaveGroupRequest and logs the {@code leaveReason}, unless this
member is using static membership or is already
- * not part of the group (ie does not have a valid member id, is in the
UNJOINED state, or the coordinator is unknown).
+ * Sends LeaveGroupRequest and logs the {@code leaveReason}, unless this
member is using static membership
+ * with the default consumer group membership operation, or is already not
part of the group (i.e., does not have a
+ * valid member ID, is in the UNJOINED state, or the coordinator is
unknown).
*
+ * @param membershipOperation the operation on consumer group membership
that the consumer will perform when closing
* @param leaveReason the reason to leave the group for logging
* @throws KafkaException if the rebalance callback throws exception
*/
- public synchronized RequestFuture<Void> maybeLeaveGroup(String
leaveReason) {
+ public synchronized RequestFuture<Void>
maybeLeaveGroup(CloseOptions.GroupMembershipOperation membershipOperation,
String leaveReason) {
RequestFuture<Void> future = null;
- // Starting from 2.3, only dynamic members will send LeaveGroupRequest
to the broker,
- // consumer with valid group.instance.id is viewed as static member
that never sends LeaveGroup,
- // and the membership expiration is only controlled by session timeout.
- if (isDynamicMember() && !coordinatorUnknown() &&
- state != MemberState.UNJOINED && generation.hasMemberId()) {
- // this is a minimal effort attempt to leave the group. we do not
- // attempt any resending if the request fails or times out.
+ if (rebalanceConfig.leaveGroupOnClose &&
shouldSendLeaveGroupRequest(membershipOperation)) {
log.info("Member {} sending LeaveGroup request to coordinator {}
due to {}",
generation.memberId, coordinator, leaveReason);
+
LeaveGroupRequest.Builder request = new LeaveGroupRequest.Builder(
rebalanceConfig.groupId,
- Collections.singletonList(new
MemberIdentity().setMemberId(generation.memberId).setReason(JoinGroupRequest.maybeTruncateReason(leaveReason)))
+ List.of(new
MemberIdentity().setMemberId(generation.memberId).setReason(JoinGroupRequest.maybeTruncateReason(leaveReason)))
Review Comment:
Just my two cents.
The client module's minimum JDK support is Java 11, so I think it's better
to leverage the modern API.
But this is not related to the topic of this patch, I will revert it in next
commit.
--
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
