sanghyeok An created KAFKA-21279:
------------------------------------
Summary: Consumer and Streams group leave append failures can lose
session timers and stall reassignment
Key: KAFKA-21279
URL: https://issues.apache.org/jira/browse/KAFKA-21279
Project: Kafka
Issue Type: Bug
Reporter: sanghyeok An
Assignee: sanghyeok An
When a member using {{group.protocol=consumer}} or {{group.protocol=streams}}
leaves its group, the coordinator cancels the member’s session timer before
persisting the member removal records.
If the subsequent local append to {{__consumer_offsets}} fails, the
coordinator’s snapshot rollback restores the member metadata, existing
assignment, and ownership state. However, session timers are not managed by the
snapshot mechanism, so the canceled timer is not restored.
As a result, the coordinator retains a member that has already shut down as the
owner of Consumer partitions or Streams active tasks, but no longer has a
session timer to expire that member. Even after the append failure is resolved
and other members continue sending successful heartbeats, the departed member’s
partitions or tasks may remain unreleased, leaving processing stalled.
The two protocols use different leave handlers, but share the same underlying
cause: timers are canceled before persistence and are not restored on rollback.
*Scenario*
# Dynamic members A and B belong to the same group, with stable assignments.
In a Consumer group, A consumes P0 and B consumes P1. In a Streams group, A and
B belong to different processes: A processes active task {{{}0_0{}}}, and B
processes active task {{{}0_1{}}}.
# A shuts down and sends a heartbeat with {{memberEpoch=-1}} using the
corresponding protocol. A sends no further heartbeats. For Streams, this is an
ordinary member departure without requesting application-wide shutdown.
# The coordinator generates A’s removal records and a group epoch update
record, then calls {{cancelTimers()}} to cancel A’s timers before persisting
the records.
# The local append of the batch containing these records fails. The
coordinator remains ACTIVE without reloading and performs a snapshot rollback.
# A’s member metadata and existing partition or active task ownership are
restored, but A’s session timer is not.
# The cause of the append failure is resolved. B continues sending successful
heartbeats and processing its existing partitions or tasks, but B’s heartbeats
refresh only B’s own session timer.
# A is not removed even after its session timeout has elapsed. Consumption of
P0 remains stalled in the Consumer group, or processing of task {{0_0}} remains
stalled in the Streams group.
Even if a subsequent rebalance changes the target assignment to move the
affected partition or task to B, B may be unable to acquire it while the
coordinator still retains A’s current ownership.
For Streams, this scenario does not require a stateful topology, standby or
warm-up replicas, or a custom assignor.
*Conditions that can cause the append failure*
One possible path is a reduction of {{max.message.bytes}} for
{{__consumer_offsets}} between batch creation and append. Both group types use
the same coordinator runtime persistence path, so this condition applies to
both.
The coordinator runtime uses the size limit in effect when creating a batch. If
the limit is subsequently reduced and the actual size of the batch containing
A’s leave records exceeds the new limit, the log append can fail with
{{{}RecordTooLargeException{}}}.
In this failure path, rolling back the batch does not necessarily reload the
coordinator. Restoring the size limit allows subsequent writes to succeed, but
does not restore A’s canceled session timer.
*Impact*
In Consumer groups, consumption of the departed member’s partitions may remain
stalled. In Streams groups, input processing for the departed member’s active
tasks may remain stalled. Other members can continue processing the partitions
or tasks they already own, so the issue does not necessarily stop processing
across the entire group.
Reloading the coordinator can register the members’ session timers again. This
is therefore not permanent log corruption, but restoring successful writes and
waiting for the session timeout are insufficient to guarantee that the affected
partitions or tasks will be released.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
