[
https://issues.apache.org/jira/browse/KAFKA-20984?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gavin Wang resolved KAFKA-20984.
--------------------------------
Resolution: Duplicate
> Static consumer can be fenced by its own abandoned JoinGroup and crash with
> FencedInstanceIdException in classic protocol
> --------------------------------------------------------------------------------------------------------------------------
>
> Key: KAFKA-20984
> URL: https://issues.apache.org/jira/browse/KAFKA-20984
> Project: Kafka
> Issue Type: Bug
> Components: clients
> Reporter: Gavin Wang
> Priority: Major
>
> A static membership consumer on the classic protocol can be fenced by its own
> abandoned `JoinGroup` request and crash with `FencedInstanceIdException`, even
> though no duplicate instance exists. We hit this on a long-running Kafka
> Streams
> application (classic protocol, static membership, EOS) when a group
> coordinator
> stall delayed request processing by about 15 seconds: both application
> instances
> received the exception on `JoinGroup`, and Streams treats it as fatal
> (`SHUTDOWN_APPLICATION`). The client behavior involved dates back to the
> introduction of static membership (KAFKA-8500). KAFKA-9659 fixed the same
> self-fencing shape, but only for the offset commit path.
> Failure sequence:
> 1. A static membership consumer calls `unsubscribe()`, which calls
> `AbstractCoordinator.maybeLeaveGroup()`. It recognizes the consumer is static
> and skips the `LeaveGroup` RPC, but `resetGenerationOnLeaveGroup()` is called
> regardless (*) and clears the local member id:
> ```java
> if (shouldSendLeaveGroupRequest(membershipOperation)) {
> log.info("Member {} sending LeaveGroup request to coordinator {} due to
> {}",
> generation.memberId, coordinator, leaveReason);
> ...
> }
> resetGenerationOnLeaveGroup(); // (*) runs even when no LeaveGroup was sent
> private boolean
> shouldSendLeaveGroupRequest(CloseOptions.GroupMembershipOperation
> membershipOperation) {
> if (!coordinatorUnknown() && state != MemberState.UNJOINED &&
> generation.hasMemberId()) {
> return membershipOperation == LEAVE_GROUP || (isDynamicMember() &&
> membershipOperation == DEFAULT);
> } else {
> return false;
> }
> }
> ```
> Every subsequent rejoin therefore carries `UNKNOWN_MEMBER_ID`, while the
> coordinator still has the old member id registered for this
> `group.instance.id`.
> 2. The consumer rejoins immediately via `JoinGroup` with `UNKNOWN_MEMBER_ID`.
> Call it Join #1.
> 3. The coordinator is stalled at this point (in our case its request pipeline
> was ~15 seconds deep while the `__consumer_offsets` partition was reloading
> after a leadership change). On the consumer's coordinator connection,
> heartbeat
> HB1 is already sitting in the coordinator's request queue, and a later
> heartbeat HB2 has been delivered but not yet read, because the broker keeps at
> most one request per connection in flight and does not read the socket again
> until the in-flight request is answered. Join #1's bytes land in the
> connection's receive buffer behind HB2, unread. HB1 eventually returns
> `COORDINATOR_NOT_AVAILABLE`, so the client marks the coordinator unknown and
> closes the connection, cancelling HB2 and Join #1 in its own bookkeeping only.
> The close cannot take back bytes already delivered: the broker will still read
> and process them whenever reading resumes. The client then rediscovers the
> same
> coordinator, opens a new connection, and sends Join #2 (still
> `UNKNOWN_MEMBER_ID`, since the id was wiped in step 1).
> 4. Join #2 arrives on the fresh connection with nothing in front of it, so it
> is read and enqueued immediately, roughly 15 seconds ahead of Join #1, which
> is
> still unread bytes (the queue orders by read time, not send time). When the
> stall clears, the backlog drains: HB2 executes (its response goes to the
> closed
> socket and is dropped), which resumes reading on the old connection, and Join
> #1
> is finally parsed and enqueued behind Join #2. Join #2 executes first: unknown
> member id plus known `group.instance.id` is handled as a static member
> replacement (`classicGroupJoinNewStaticMember` ->
> `updateStaticMemberThenRebalanceOrCompleteJoin` ->
> `ClassicGroup.replaceStaticMember`), and its response is held while the new
> membership record replicates. Then Join #1 executes carrying the exact same
> signature. The coordinator cannot distinguish a retry by the same consumer
> from
> a second process claiming the instance id, so it applies the replacement logic
> again: it evicts Join #2's member, and `replaceStaticMember` completes Join
> #2's
> pending response with `FENCED_INSTANCE_ID`:
> ```java
> // Fence potential duplicate member immediately if someone awaits join/sync
> future.
> JoinGroupResponseData joinGroupResponse = new JoinGroupResponseData()
> ...
> .setErrorCode(Errors.FENCED_INSTANCE_ID.code());
> completeJoinFuture(removedMember, joinGroupResponse);
> ```
> Join #2's connection is still open on the client side, so this error is
> actually delivered; the consumer throws `FencedInstanceIdException`, which is
> treated as fatal. Join #1's own success response is dropped on the closed
> connection.
> The problem: fencing the earlier pending attempt is correct when two distinct
> processes really share a `group.instance.id`. The bug is that (*) makes a lone
> consumer produce that exact signature against itself: clearing the member id
> without having sent `LeaveGroup` makes every rejoin indistinguishable from a
> duplicate process, so any situation that leaves one `JoinGroup` abandoned but
> deliverable (a coordinator stall plus a client-side disconnect, in our case)
> lets the abandoned request fence the retry.
> Proposed fix: when `maybeLeaveGroup()` suppresses the `LeaveGroup` RPC for a
> static member, keep the member id instead of resetting it. Both joins then
> identify as the existing member and are handled as a rejoin rather than a
> replacement, so no ordering of connections and queues can produce a
> self-fence.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
