Manages membership in a Kafka consumer group.

This module implements the Kafka consumer group protocol, handling:

• Group Membership: Joining and leaving consumer groups via JoinGroup/LeaveGroup requests
• Partition Assignment: Coordinating partition distribution among group members
• Rebalancing: Triggering rebalances when members join/leave or partitions change
• Heartbeats: Maintaining group membership through periodic heartbeat messages

Consumer Group Protocol Flow

1. Join Phase: Send JoinGroupRequest to group coordinator. The coordinator blocks until all members have joined, then elects a leader.

2. Sync Phase: The leader computes partition assignments and sends them via SyncGroupRequest. All members receive their assignments in the response.

3. Consume Phase: Members consume from assigned partitions while sending periodic heartbeats to maintain membership.

4. Rebalance: When the coordinator signals a rebalance (via heartbeat response), members stop consuming, commit offsets, and rejoin the group.

Options

The following options can be passed when starting a consumer group:

• :heartbeat_interval - Interval between heartbeats in ms (default: 3000)
• :session_timeout - Session timeout in ms (default: 45000)
• :session_timeout_padding - Extra time added to request timeouts (default: 10000)
• :rebalance_timeout - Time allowed for consumers to rejoin during rebalance (default: session_timeout * 3)
• :partition_assignment_callback - Function for custom partition assignment (default: round-robin)
• :crash_rejoin_max_restarts - Abnormal heartbeat-crash rejoins tolerated within crash_rejoin_window_ms before a terminal stop (default: 10; :infinity disables the bound)
• :crash_rejoin_window_ms - Sliding window (ms) for crash_rejoin_max_restarts (default: 60000)

Options can also be configured globally via application config under :kafka_ex.

Crash-loop bound and recovery

This bound governs only an abnormal, uncatchable heartbeat death — an :kill or an unstructured process crash that reaches the manager as a bare {:EXIT, _, reason} from the current heartbeat. Every catchable heartbeat failure is already classified by the heartbeat process itself (recoverable errors → rejoin; a code exception or :fenced_instance_id / :group_authorization_failed → immediate terminal stop), so those do not go through this counter.

Such an abnormal death triggers an in-place rejoin (and a [:kafka_ex, :consumer, :heartbeat_crash] event). If they keep happening — more than crash_rejoin_max_restarts times within crash_rejoin_window_ms (defaults 10 / 60_000 ms; OTP max_restarts/max_seconds semantics) — the member stops terminally, emits [:kafka_ex, :consumer, :member_terminated] with {:crash_loop, reason}, and is torn down by its one_for_all / max_restarts: 0 supervisor (whose max_restarts: 0 is intentional and load-bearing for this escalation). The window is a pure sliding time-window: it is never reset on a successful rejoin, only aged out, so crashes are counted across the member's whole lifetime within the window. Prior to this change the loop was unbounded; the bound is active by default.

To get automatic recovery from a terminal stop, run ConsumerGroup under your own supervisor (the common case); it is then restarted fresh under your restart strategy. A ConsumerGroup started standalone will simply stop (see UPGRADING.md). A correlated kill wave (e.g. a fleet-wide OOM or deploy) can trip many members at nearly the same time — per-member windowing limits each member, but jitter does NOT desync the trip itself, so size your own supervisor's max_restarts/max_seconds to absorb a simultaneous restart; it is the real backstop. The bound counts one crash per heartbeat failure, so with a short heartbeat_interval crashes accumulate faster within the window — raise crash_rejoin_max_restarts to give more headroom before tripping. Set crash_rejoin_max_restarts: :infinity to disable the bound.