[
https://issues.apache.org/jira/browse/KAFKA-21279?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias J. Sax updated KAFKA-21279:
------------------------------------
Labels: (was: group-coordinator)
> 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
> Priority: Major
>
> 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)
