Consumers and Consumer Groups

After assign(), the consumer starts from the last committed offset for that group.id if one exists; otherwise auto.offset.reset (earliest/latest) applies. You can override explicitly with seek(), seekToBeginning(), or seekToEnd() before polling.

subscribe(Pattern) matches topic names against a regex. The consumer auto-discovers new matching topics and rebalances to include them. It still uses group management, just with a dynamic topic set.

fetch.min.bytes tells the broker the minimum data to accumulate before answering a fetch; the broker waits up to fetch.max.wait.ms for that much to arrive, then responds anyway. Bigger min.bytes = better throughput but higher latency.

position() is where the consumer will read next (in-memory). committed() is the last offset persisted to Kafka for the group. beginningOffsets()/endOffsets() are the oldest and newest (end) offsets currently in each partition.

KafkaAvroDeserializer reads a schema-registry ID from each record and fetches the writer schema. With specific.avro.reader=false (default) you get a generic GenericRecord; with true you get generated, typed SpecificRecord classes.

The generation ID is a monotonically increasing counter the coordinator bumps on each successful rebalance; member.id uniquely identifies a member within the group. Requests carrying a stale generation or unknown member.id are rejected, fencing zombies.

The coordinator's state and committed offsets live in __consumer_offsets. If its replication factor is too low (e.g. 1), losing that broker loses group/offset state and breaks coordinator failover, causing duplicate processing or stuck groups.

They are broker-side limits on the session.timeout.ms a consumer may request: group.min.session.timeout.ms (default 6000 ms) and group.max.session.timeout.ms (default 1800000 ms). A consumer asking for a value outside this range is rejected and cannot join the group.

session.timeout.ms detects a crashed or network-isolated consumer via missed heartbeats from the background thread. max.poll.interval.ms detects a consumer that is alive but stuck/slow in processing because it hasn't called poll() in time. Before KIP-62 both were conflated into the single session timeout.

Dedicated monitors continuously read every group's committed offsets and partition LEOs externally, so they see lag even when consumers are dead, and expose it to Prometheus/alerting. Burrow adds a threshold-free status evaluation; Lag Exporter adds time-based lag estimates.

__consumer_offsets is an internal Kafka topic (50 partitions by default) where consumer commits are stored as messages keyed by (group, topic, partition). It's log-compacted so only the latest committed offset per key is retained. The group coordinator broker reads/writes it to track group progress.

Register a ConsumerRebalanceListener via subscribe(). In onPartitionsRevoked, commit offsets for partitions you're about to lose before they move to another consumer. onPartitionsAssigned runs when you gain partitions (e.g., to seek to a custom position). This prevents duplicate or lost processing across rebalances.

max.partition.fetch.bytes caps bytes returned per partition per fetch (default ~1MB); fetch.max.bytes caps total bytes across all partitions (~50MB). Both are soft limits: to avoid stalling, a broker always returns at least one full record batch even if it exceeds the cap.

The consumer keeps a buffer of records per partition and issues fetch requests for more data BEFORE the buffer is empty (prefetch). So while you process the current batch, the next fetch is already in flight, and poll() usually returns instantly from the buffer instead of waiting on the network.

Use offsetsForTimes(map of partition to timestamp) to look up the first offset whose record timestamp is >= the given time, then seek() to it. If no record has a timestamp at or after the target, it returns null for that partition.

ConsumerRebalanceListener has onPartitionsRevoked, onPartitionsAssigned, and onPartitionsLost. Commit offsets in onPartitionsRevoked before giving partitions up. Under eager you're given all partitions; under cooperative you only get the ones actually moving.

If two live consumers join the same group with the same group.instance.id, the broker treats them as duplicates of one static member and fences the older one, throwing FencedInstanceIdException so only one instance keeps consuming each set of partitions.

Give each instance a stable group.instance.id and set session.timeout.ms longer than a single instance's worst-case restart time. Then rolling-restart one instance at a time so each returns inside its window — no rebalances. The trade-off: a longer timeout slows detection of genuinely crashed instances.

If a deserializer instantiates arbitrary Java classes named in the payload (e.g., via type headers or Java native serialization), an attacker who controls the bytes can trigger remote code execution. Mitigate by restricting allowed types — e.g., Spring's JsonDeserializer trusted.packages allow-list.

Skew shows as lag concentrated on a few partitions while others sit at zero — usually hot keys or a stuck/slow consumer. For autoscaling, scale consumers on total/time lag, but cap replicas at the partition count, throttle on rebalances, and prefer time-lag with hysteresis to avoid flapping.

Match the lever to the cost model: lower max.poll.records if cost scales with count; offload with pause()/resume() if individual records are slow or variable; raise max.poll.interval.ms only for legitimately long, bounded work. Cranking the interval high delays detection of truly stuck consumers, slowing partition reassignment and recovery.

Use assign() to pin a fixed partition set to each instance based on a stable ordinal (e.g. pod index). You avoid rebalances and get deterministic ownership, but you give up automatic load balancing and failover — you must handle scaling and dead-instance takeover yourself.

Use the kafka-consumer-groups CLI --reset-offsets (to earliest, a timestamp, or a specific offset) while the group is stopped, or seek() the partitions in code. auto.offset.reset can't do it because committed offsets already exist.

Switch to CooperativeStickyAssignor so most partitions keep running during rebalances, enable static membership (group.instance.id) so restarts don't trigger rebalances, and tune session.timeout.ms / heartbeat.interval.ms / max.poll.interval.ms so consumers aren't falsely evicted.

Static membership (group.instance.id) avoids rebalances entirely for transient restarts. Cooperative-sticky rebalancing makes rebalances that DO happen incremental — only affected partitions move instead of stopping the whole group. They solve different problems and are best used together.