Consumer offsets

Deep dive

How it works

In: group, partition, last processed offset · Out: committed offset to resume from · State: one integer (plus epoch, metadata) per (group, partition), compacted · Cost: O(1) per commit and lookup; one replicated append per commit request

1. The group's coordinator owns the commit state for the group; a new coordinator (after failover) rebuilds it by replaying the partition's log.
2. A consumer fetches from position, processes the batch, then sends OffsetCommit(group, generation, member, [(topic, partition, last+1)]).
3. The coordinator rejects commits whose generation is older than the group's current one. After a rebalance a zombie therefore cannot overwrite the new owner's progress (a fencing token).
4. The commit is appended as a record, key (group, topic, partition), value (offset, leader epoch, metadata, timestamp), replicated like any other log record, then applied to the in-memory table and acked.
5. Compaction keeps only the newest record per key, so size is bounded by the number of keys, not the number of commits. A tombstone (null value) deletes an entry (expiry, group deletion).
6. On start or after a rebalance, the consumer asks for the committed offset; none found means a policy (auto.offset.reset: earliest, latest, error).

poll ──► records [100..109] ──► process ──► commit(110)
                      crash here ▲  (before commit) → restart at 100, 100..109 again
partition log of __consumer_offsets:  (g,t,0)=90 (g,t,1)=40 (g,t,0)=100 (g,t,0)=110
after compaction:                      (g,t,1)=40 (g,t,0)=110

Code

Coordinator-side store: append-only byte log, in-memory index, generation fencing, recovery by replay with a torn-tail check, compaction. Compiled and tested (at-least-once replay after a crash, a fenced zombie, torn tail, compaction).

use std::collections::HashMap;

pub type Key = (String, String, u32); // (group, topic, partition)

#[derive(Debug, PartialEq)]
pub enum CommitError { StaleGeneration { current: u32 } }

/// Group coordinator state: an append-only commit log plus an in-memory index.
/// The stored offset is the NEXT offset to read (last processed + 1).
#[derive(Default)]
pub struct OffsetStore {
    pub log: Vec<u8>,                 // stands in for the compacted __consumer_offsets log
    index: HashMap<Key, u64>,         // last record per key wins
    generation: HashMap<String, u32>, // bumped on every rebalance
}

// record = len:u32 | partition:u32 | has_offset:u8 | offset:u64 | glen:u16 group | tlen:u16 topic
fn encode(k: &Key, off: Option<u64>) -> Vec<u8> {
    let mut b = Vec::with_capacity(21 + k.0.len() + k.1.len());
    b.extend_from_slice(&[0; 4]); // length placeholder
    b.extend_from_slice(&k.2.to_le_bytes());
    b.push(off.is_some() as u8);
    b.extend_from_slice(&off.unwrap_or(0).to_le_bytes());
    for s in [&k.0, &k.1] {
        b.extend_from_slice(&(s.len() as u16).to_le_bytes());
        b.extend_from_slice(s.as_bytes());
    }
    let len = (b.len() - 4) as u32;
    b[..4].copy_from_slice(&len.to_le_bytes());
    b
}

fn decode(buf: &[u8]) -> Option<(Key, Option<u64>, usize)> {
    let len = u32::from_le_bytes(buf.get(..4)?.try_into().ok()?) as usize;
    let body = buf.get(4..4 + len)?; // None on a torn tail
    let partition = u32::from_le_bytes(body.get(..4)?.try_into().ok()?);
    let off = u64::from_le_bytes(body.get(5..13)?.try_into().ok()?);
    let (g, rest) = take_str(body.get(13..)?)?;
    let (t, _) = take_str(rest)?;
    Some(((g, t, partition), if body[4] == 1 { Some(off) } else { None }, 4 + len))
}

fn take_str(b: &[u8]) -> Option<(String, &[u8])> {
    let n = u16::from_le_bytes(b.get(..2)?.try_into().ok()?) as usize;
    let s = String::from_utf8(b.get(2..2 + n)?.to_vec()).ok()?;
    Some((s, &b[2 + n..]))
}

impl OffsetStore {
    /// Recover by replaying the log; a torn final record is dropped.
    pub fn open(log: &[u8]) -> Self {
        let mut s = OffsetStore::default();
        let mut pos = 0;
        while let Some((k, off, n)) = decode(&log[pos..]) {
            match off { Some(o) => { s.index.insert(k, o); } None => { s.index.remove(&k); } }
            pos += n;
        }
        s.log = log[..pos].to_vec();
        s
    }

    pub fn rebalance(&mut self, group: &str) -> u32 {
        let g = self.generation.entry(group.to_string()).or_insert(0);
        *g += 1;
        *g
    }

    /// Fenced commit: a zombie from an older generation is rejected.
    pub fn commit(&mut self, group: &str, generation: u32, topic: &str, partition: u32, next: u64) -> Result<(), CommitError> {
        let current = *self.generation.get(group).unwrap_or(&0);
        if generation < current { return Err(CommitError::StaleGeneration { current }); }
        let k = (group.to_string(), topic.to_string(), partition);
        self.log.extend_from_slice(&encode(&k, Some(next))); // real systems fsync/replicate before acking
        self.index.insert(k, next);
        Ok(())
    }

    pub fn committed(&self, group: &str, topic: &str, partition: u32) -> Option<u64> {
        self.index.get(&(group.to_string(), topic.to_string(), partition)).copied()
    }

    /// Log compaction: keep only the latest record per key.
    pub fn compact(&self) -> Vec<u8> {
        let mut keys: Vec<_> = self.index.iter().collect();
        keys.sort();
        keys.into_iter().flat_map(|(k, o)| encode(k, Some(*o))).collect()
    }
}

The consumer side is a loop: fetch(position) → process → commit(position') with the commit as the last step. Fewer commits, more duplicates on crash; the bound is the records since the last commit.

Complexity and constants

• Commit and lookup: O(1) (hash table); one record is tens of bytes (this sketch: 21 bytes plus names; Kafka adds epoch, metadata and timestamp, order of 50-100 bytes).
• Storage: O(distinct keys) after compaction; 100,000 keys at 50-100 bytes is order of 5-10 MB.
• Request rate = consumers x partitions-per-request / commit interval. Auto-commit interval defaults to 5 s, so 10,000 consumers send order of 2,000 commit requests/s, each one durable write on the coordinator.
• The cost driver is the replicated, ack-waiting append, not the bytes. One request carries all of a consumer's partitions, and the coordinator batches appends from many consumers (group commit).

Efficiency

• Cost: one commit is one replicated, ack-waiting append of a record of order of 50-100 bytes (Kafka); the lookup is a hash-table read. The bound is write latency and commit request rate, not bytes: 10,000 consumers at the 5 s default interval send order of 2,000 commit requests/s. Compacted state is about 5-10 MB per 100,000 keys (order of).
• Levers:
  • commit per fetched batch, not per record — cuts replicated writes by the batch size — any consumer
  • longer commit interval — request rate falls in proportion — loads that tolerate more replay after a crash
  • one request carries all of a consumer's partitions — fewer round trips — wide assignments
  • coordinator batches appends from many consumers (group commit) — one fsync/replication round per batch — large groups
  • small log segments plus an in-memory index — fast replay and compaction on failover (offsets.topic.segment.bytes 100 MiB default) — coordinator moves
• Naive → standard → state of the art:
  • commit every record · one replicated write per message
  • commit per batch or every 5 s · order of 2,000 requests/s at 10,000 consumers
  • offset stored in the same transaction as the result, or checkpointed with state (Flink) · no separate commit round trip on the hot path

Implementation notes

• Commit once per fetched batch or per interval, never per record. A commit per record turns each message into a replicated write.
• Kafka's auto-commit (enable.auto.commit=true, 5 s) commits inside poll() the offsets returned by the previous poll(): safe (at-least-once) only if the whole batch is processed before the next poll().
• Async commit has no ordering guarantee across retries: an older commit retried after a newer one would move the offset back. Use a sequence counter, or one final sync commit on shutdown.
• Keep the offset index in memory on the coordinator and load it by sequential replay of one compact partition; keep the offsets log segments small so replay and compaction stay fast (Kafka default offsets.topic.segment.bytes 100 MiB).
• Never change the number of offsets-topic partitions on a live cluster: the group-to-partition hash would move groups.
• Store offsets next to your results (same DB transaction) when you need exactly-once effects. The broker's copy is then only for monitoring.

Pitfalls

• Committing the processed offset instead of processed + 1 replays one record on every restart.
• Committing before processing and then failing loses records silently.
• Committing an offset acknowledges everything below it. Processing a batch in parallel and committing the highest finished offset loses records still in flight. Track the lowest contiguous finished offset.
• Expired offsets (group idle past retention) fall back to auto.offset.reset: latest silently skips data, earliest replays everything.
• Offsets pointing below the log start (retention deleted the data) trigger the same reset policy.
• A commit during a rebalance, from a member that lost its partition, is the zombie case: it must fail, and the consumer must not retry it.
• Committing after a slow processing step longer than the session timeout gets the member kicked out and the commit rejected.

Variants and alternatives

In production

Messaging: Apache Kafka and Kafka-API compatible brokers (Redpanda) store group offsets in the internal offsets topic; Pulsar tracks a per-subscription cursor in BookKeeper; NATS JetStream keeps per-consumer ack state. Streaming: Amazon Kinesis consumers checkpoint their position in DynamoDB through the Kinesis Client Library. Databases: Postgres logical replication slots remember how far a change-data-capture reader has confirmed (confirmed_flush_lsn), and Debezium stores its own source offsets. Older Kafka consumers kept offsets in ZooKeeper, which stops scaling at high commit rates.

Implementations

Sources

• Apache Kafka documentation, broker and consumer configuration (offsets.*, enable.auto.commit, auto.commit.interval.ms). docs
• KIP-211: Revise Expiration Semantics of Consumer Group Offsets. KIP
• KIP-98: Exactly Once Delivery and Transactional Messaging. KIP
• KIP-848: The Next Generation of the Consumer Rebalance Protocol. KIP
• Jay Kreps, Neha Narkhede, Jun Rao, Kafka: a Distributed Messaging System for Log Processing, NetDB 2011.
• PostgreSQL documentation, Logical Decoding Concepts (replication slots track the consumer's confirmed position). docs
• AWS, KCL 1.x and 2.x information (checkpointing, DynamoDB lease table), Kinesis Data Streams Developer Guide. docs
• Apache Pulsar documentation, messaging concepts (acknowledgment, cursors). docs