[jira] [Created] (KAFKA-21279) Consumer and Streams group leave append failures can lose session timers and stall reassignment

["sanghyeok An (Jira)" <ji...@apache.org>]

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)