frankvicky commented on code in PR #23266:
URL: https://github.com/apache/kafka/pull/23266#discussion_r3913400272
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##########
@@ -1180,7 +1177,13 @@ public synchronized RequestFuture<Void>
maybeLeaveGroup(CloseOptions.GroupMember
client.pollNoWakeup();
}
- resetGenerationOnLeaveGroup();
+ // A static member whose LeaveGroup is suppressed stays registered
with the group
+ // coordinator under its current member id, so keep the id locally as
well: resetting
+ // it would make the next rejoin carry UNKNOWN_MEMBER_ID, which the
coordinator must
+ // treat as a new instance claiming this group.instance.id, fencing
any still-pending
+ // join attempt of this same consumer (KAFKA-20985). State and
generation are reset
+ // and a rejoin is requested exactly as before.
+ resetStateAndRejoin("consumer pro-actively leaving the group",
shouldSendLeaveGroup || isDynamicMember());
Review Comment:
One behavioral change that I think should be called out in the PR
description: a static *leader* that rejoins while the group is `STABLE` now
goes through `classicGroupJoinExistingMember`, which always forces a rebalance
when the leader sends JoinGroup. The previous static-replacement path
(`updateStaticMemberThenRebalanceOrCompleteJoin`) skipped the rebalance as long
as the selected protocol didn't change.
So after a poll timeout or an `unsubscribe()`/`subscribe()` cycle (e.g.
`StreamThread.handleTaskMigrated`), a static leader will now trigger a full
group rebalance where it previously wouldn't.
I think this is acceptable (the old path also ignored subscription changes,
so the new one is arguably more correct), but please mention it in the
description so it's visible from the commit history.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##########
@@ -1180,7 +1177,13 @@ public synchronized RequestFuture<Void>
maybeLeaveGroup(CloseOptions.GroupMember
client.pollNoWakeup();
}
- resetGenerationOnLeaveGroup();
+ // A static member whose LeaveGroup is suppressed stays registered
with the group
Review Comment:
nit: this comment could be trimmed to two or three lines. The full failure
sequence is already in the JIRA and the PR description; here it's enough to say
that a static member whose LeaveGroup was suppressed is still registered under
this member id, so we keep it to avoid being treated as a new instance on the
next rejoin.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java:
##########
@@ -1180,7 +1177,13 @@ public synchronized RequestFuture<Void>
maybeLeaveGroup(CloseOptions.GroupMember
client.pollNoWakeup();
}
- resetGenerationOnLeaveGroup();
+ // A static member whose LeaveGroup is suppressed stays registered
with the group
+ // coordinator under its current member id, so keep the id locally as
well: resetting
+ // it would make the next rejoin carry UNKNOWN_MEMBER_ID, which the
coordinator must
+ // treat as a new instance claiming this group.instance.id, fencing
any still-pending
+ // join attempt of this same consumer (KAFKA-20985). State and
generation are reset
+ // and a rejoin is requested exactly as before.
+ resetStateAndRejoin("consumer pro-actively leaving the group",
shouldSendLeaveGroup || isDynamicMember());
Review Comment:
There is still a residual (non-fatal) race with this fix that I think is
worth recording in the JIRA: if both joins are processed while the group is in
`PREPARING_REBALANCE`, the second one goes through
`updateMemberThenRebalanceOrCompleteJoin` -> `ClassicGroup.updateMember`, which
overwrites the member's pending join future without completing it. If the join
from the surviving connection is the one that gets overwritten, the consumer
never receives a response and waits for the JoinGroup request timeout
(rebalance timeout + 5s) before retrying.
That is much better than being fenced, so no change needed in this PR, just
worth documenting as a known limitation / possible follow-up.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinatorTest.java:
##########
@@ -1142,6 +1143,33 @@ private static Stream<Arguments>
groupInstanceIdAndMembershipOperationMatrix() {
);
}
+ @Test
+ public void testStaticMemberKeepsMemberIdWhenLeaveGroupIsSuppressed() {
Review Comment:
Could we also cover `REMAIN_IN_GROUP` here? For a static member it is the
other case where no LeaveGroup is sent and the member id should now be kept.
Turning this into a `@ParameterizedTest` over `DEFAULT` and `REMAIN_IN_GROUP`
would be enough.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
