Gavin Wang created KAFKA-20983:
----------------------------------
Summary: Static consumer can be fenced by its own abandoned
JoinGroup and crash with FencedInstanceIdException in classic protocol
Key: KAFKA-20983
URL: https://issues.apache.org/jira/browse/KAFKA-20983
Project: Kafka
Issue Type: Bug
Components: clients
Reporter: Gavin Wang
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)
