Consumers and Consumer Groups

The Group Coordinator is a broker that manages a consumer group: it handles members joining/leaving, triggers rebalances, and stores offsets. The coordinator is the broker that leads the __consumer_offsets partition the group maps to.

A heartbeat is a small periodic signal a consumer sends to the group coordinator broker to prove it is alive and still part of its consumer group. If heartbeats stop, the broker assumes the consumer died and reassigns its partitions.

Run kafka-consumer-groups.sh --bootstrap-server host:port --describe --group <group>. It lists each partition with CURRENT-OFFSET (committed), LOG-END-OFFSET (latest), and LAG = the difference, plus the owning consumer.

Consumer lag is how far behind a consumer is on a partition: the latest message offset (log-end-offset) minus the offset the consumer has committed. Lag of 0 means fully caught up.

max.poll.interval.ms is the maximum time allowed between two poll() calls. If processing a batch takes longer, Kafka assumes the consumer is stuck, kicks it out of the group, and rebalances its partitions to other members.

subscribe() joins a consumer group and lets Kafka assign partitions automatically (dynamic, rebalances on membership change). assign() manually pins specific partitions to the consumer with no group coordination or rebalancing.

A committed offset records how far a consumer group has processed in a partition. It stores the offset of the NEXT message to read (last processed + 1), so a restart resumes from there without re-reading.

poll(Duration) fetches a batch of records the consumer already has buffered (or waits up to the timeout for some), and also drives heartbeats, group coordination, and offset commits. You loop because all consumer work happens during poll.

auto.offset.reset decides where a consumer starts reading when there is no valid committed offset for a partition. earliest starts at the oldest message, latest at the newest, and none throws an exception.

A rebalance is when Kafka redistributes a topic's partitions among the consumers in a group. It is triggered when a consumer joins or leaves the group, or when the set of subscribed partitions changes.

Static membership gives a consumer a stable identity via the group.instance.id config (KIP-345). With it, a brief consumer restart does not trigger a group rebalance, because the broker recognizes the returning member by its fixed id.

group.id names the consumer group that shares partitions and offsets. client.id is a logical label for the connection used in logs/metrics/quotas. isolation.level (read_uncommitted or read_committed) decides whether the consumer sees records from uncommitted transactions.

All members send JoinGroup; the coordinator picks one member as group leader and returns the full member list to it. The leader computes assignments and sends them in SyncGroup; the coordinator fans them out to each member in their SyncGroup response.

heartbeat.interval.ms is how often the consumer sends heartbeats; session.timeout.ms is how long the coordinator waits without one before declaring the consumer dead. The interval must be well below the timeout — the rule of thumb is heartbeat.interval.ms <= session.timeout.ms / 3.

records-lag-max is a client-side JMX metric exposed by each consumer reporting the maximum lag across the partitions it currently fetches. It's cheap and real-time but only sees assigned partitions of running consumers.

With enable.auto.commit=true, the consumer commits the current position periodically (auto.commit.interval.ms, default 5000ms) during poll(). It's simple but can lose or duplicate records on crash. Manual commits (commitSync/commitAsync after processing) give you control over timing and delivery semantics.

max.poll.records caps how many records a single poll() call returns to your code (default 500). Fetch-size settings (fetch.max.bytes, max.partition.fetch.bytes) control how much data is pulled from brokers over the wire — bytes, not record count.

These methods manually set where the consumer reads next. seek() jumps to a specific offset; seekToBeginning() to the oldest; seekToEnd() to the newest. The change takes effect on the next poll(), and overrides auto.offset.reset.

RangeAssignor assigns contiguous partition ranges per topic (can imbalance with multiple topics). RoundRobinAssignor spreads all partitions evenly across consumers. StickyAssignor balances evenly while trying to keep previous assignments stable across rebalances.

A static member keeps its slot during a brief restart because it doesn't send LeaveGroup and the coordinator hasn't expired its session yet. If it rejoins within session.timeout.ms with the same group.instance.id, it gets the same partitions back — no rebalance.

A poison pill is a record whose bytes can't be deserialized, so it throws on every poll and blocks the consumer forever. ErrorHandlingDeserializer wraps the real deserializer, catches the failure, and hands the error to a recoverer instead of crashing.

The coordinator tracks each member's last heartbeat. If a member sends none for session.timeout.ms, the coordinator declares it dead, removes it from the group, and starts a rebalance that reassigns the dead member's partitions to the remaining consumers.

Hand records to a worker pool for slow processing. To avoid fetching more than you can handle, call consumer.pause() on the assigned partitions and keep calling poll() (which now returns nothing but proves liveness). When workers catch up, call resume(). Commit only completed offsets.

It's a callback you pass to subscribe() so you can react to rebalances: onPartitionsRevoked (commit offsets / flush state before losing partitions), onPartitionsAssigned (init state / seek for new partitions), onPartitionsLost (partitions gone unexpectedly — don't commit).

Disable auto-commit, process the records first, then commit. Never commit before the work is durable. Because a crash between processing and commit replays the batch, make handlers idempotent. For exactly-once, use Kafka transactions or commit offsets atomically with your output to an external store.

In eager rebalancing every consumer revokes all its partitions and stops processing until a new assignment arrives (stop-the-world). Incremental cooperative rebalancing (KIP-429) only revokes the partitions that actually need to move, so most partitions keep being processed throughout.

The coordinator tracks a group's lifecycle: Empty (no members), PreparingRebalance (waiting for members to join), CompletingRebalance (awaiting SyncGroup), Stable (assignment done, consuming), and Dead (group removed). Membership changes drive transitions through these states.

session.timeout.ms / heartbeat.interval.ms detect a dead consumer via a background heartbeat thread. max.poll.interval.ms detects a live-but-stuck consumer via the poll() call. They were split (KIP-62) so slow processing doesn't get confused with a crash.

max.poll.records caps how many records each poll() returns. Lowering it means smaller batches that process faster, so poll() is called again sooner and you stay under max.poll.interval.ms. The tradeoff is more poll round-trips and potentially lower throughput.