Offset Management & Rebalancing

#Introduction

Offsets are how a consumer group remembers its progress. When you commit them decides whether a crash causes duplicates or lost messages. Rebalancing — redistributing partitions when the group changes — is the other source of surprises. Both are favourite interview topics.

#Committed Offsets

A consumer group commits, for each partition, the offset of the next record it wants to read. Commits are stored in Kafka's internal topic __consumer_offsets.

After processing offset 42, the group commits 43. If the consumer restarts, or the partition moves to another consumer, reading resumes at 43.

#When You Commit Decides the Guarantee

At-least-once — commit after processing — is the usual choice, combined with consumers that can handle the occasional duplicate (see Delivery Semantics & Exactly-Once).

#Auto-Commit

With enable.auto.commit=true (the default), the consumer commits the offsets returned by previous poll() calls every auto.commit.interval.ms (5 s) — the commit happens inside poll().

For anything important, most teams disable auto-commit and commit explicitly. (Spring Kafka does this for you — it disables Kafka's auto-commit and commits through its listener container.)

#Manual Commits

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

commitSync — Commit After Each Batch

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        process(record);
    }
    consumer.commitSync();   // commits the offsets after the last polled records; retries until success or fatal error
}

commitAsync — Non-Blocking

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) log.warn("Commit failed for {}", offsets, exception);
});

A common pattern: commitAsync() in the loop for speed, and commitSync() once on shutdown or before partitions are revoked.

try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        records.forEach(this::process);
        consumer.commitAsync();
    }
} catch (WakeupException e) {
    // shutting down
} finally {
    try {
        consumer.commitSync();   // final, reliable commit
    } finally {
        consumer.close();
    }
}

Committing Specific Offsets

To commit part-way through a large batch, pass explicit offsets — remember to commit offset + 1:

Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1));
consumer.commitSync(offsets);

#Seeking and Replaying

A consumer can move its position at any time:

consumer.seekToBeginning(consumer.assignment());
consumer.seek(new TopicPartition("orders", 2), 500);

// Jump to the first offset at or after a timestamp
Map<TopicPartition, OffsetAndTimestamp> found = consumer.offsetsForTimes(
        Map.of(new TopicPartition("orders", 2), yesterdayMillis));

Use seeking to reprocess after a bug fix, or to skip a known-bad record. For whole groups, the CLI's --reset-offsets does the same (see Running Kafka Locally & the CLI).

#Rebalancing

A rebalance redistributes partitions among the consumers in a group. It's triggered when:

• a consumer joins the group (new instance, scale-up, deployment);
• a consumer leaves cleanly (close()) or fails (missed heartbeats → session.timeout.ms);
• a consumer exceeds max.poll.interval.ms and is removed;
• the subscribed topics change (partitions added, a new topic matches a pattern subscription).

Eager vs Cooperative Rebalancing

Reacting to Rebalances

Commit what you've processed before losing a partition:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync(currentOffsets);   // don't lose progress on partitions being taken away
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // optionally seek to offsets stored elsewhere (e.g. in your database)
    }
});

Static Membership

Every deployment or restart normally triggers two rebalances (leave + rejoin). Setting group.instance.id to a stable value per instance (e.g. the Kubernetes pod name) makes the consumer a static member: a restart within session.timeout.ms gets its old partitions back without a rebalance.

props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, System.getenv("HOSTNAME"));

Kafka 4: the New Consumer Group Protocol

Kafka 4.0 made the next-generation consumer rebalance protocol (KIP-848) generally available:

props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "consumer");   // default is still "classic"

• The broker (group coordinator) computes assignments instead of a client "leader".
• Rebalances are incremental by design — no global synchronization barrier.
• Much faster and more stable rebalancing for large groups; assignors are configured on the server side (group.remote.assignor).

#Avoiding Rebalance Storms

• Keep each poll's processing well under max.poll.interval.ms (lower max.poll.records).
• Use the CooperativeStickyAssignor (or the new protocol).
• Use static membership for rolling deployments.
• Close consumers gracefully on shutdown so partitions move immediately.
• Avoid very short session.timeout.ms values in environments with GC pauses or noisy networks.

#Key Takeaways

• Committed offsets live in __consumer_offsets and mark the next record to read.
• Commit after processing for at-least-once; commit before for at-most-once.
• Disable auto-commit when processing isn't strictly synchronous; use commitAsync + a final commitSync.
• Rebalances happen on joins, leaves, failures and poll timeouts.
• Cooperative rebalancing, static membership and Kafka 4's new protocol reduce disruption.