Definition
- Committed offset -- Kafka tracks each consumer group's position per partition as a committed offset in the internal
__consumer_offsetstopic; a record is only considered done when its offset is committed. - Crash before commit -- if a consumer reads records and crashes before committing, the committed offset still points before those records, so on resume they are re-read and reprocessed.
- At-least-once delivery -- no message is lost, but a message can be processed more than once; this is the defining behavior when you commit after processing.
- Manual commit -- disable
enable.auto.commitand callcommitSync()only after processing succeeds, so a crash before commit causes a safe reprocess rather than a loss. - Idempotent processing -- since duplicates are possible, design the downstream so applying the same record twice has the same effect as once, using an idempotency key like the transfer id.
Imagine a settlement clerk at National Bank of India who takes a stack of transfer slips off the tray, starts working, but faints before marking the tray as cleared up to that slip. When a new clerk arrives, the tray still shows those slips as not yet cleared, so the new clerk picks them up again. Nothing is lost, but a slip might get worked twice, so the ledger must be designed so that applying the same transfer twice does not double-charge Rajesh. The safe rule is: only mark the tray cleared after the work is truly done.
Understanding the Concepts
What happens when a consumer crashes before committing?
Kafka tracks each consumer group's position per partition as a committed offset stored in the internal offsets topic. A record is only considered done when its offset is committed. So if a consumer polls records at offsets 50 through 59, begins processing, and crashes at offset 55 before committing, the committed offset is still 50. When the group rebalances and another member takes over, it resumes from offset 50 and reprocesses 50 through 59.
package com.bank.kafka;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class CrashBeforeCommitDemo {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "settlement-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Manual commit
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("transfers"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
for (ConsumerRecord<String, String> record : records) {
System.out.println("Processing: offset=" + record.offset()
+ " key=" + record.key() + " value=" + record.value());
// If crash happens HERE -- before commitSync below --
// the committed offset stays at the previous value
// and these records will be reprocessed on resume
}
consumer.commitSync(); // Commit AFTER processing succeeds
System.out.println("Offsets committed -- records are done");
}
}
}
Expected output:
Processing: offset=50 key=SAV2001 value=DEBIT 50000
Processing: offset=51 key=SAV3001 value=CREDIT 50000
Offsets committed -- records are done
Why is auto-commit dangerous?
With enable.auto.commit=true, the consumer commits offsets automatically every auto.commit.interval.ms, by default every five seconds. The danger is that auto-commit can commit offsets for records you have polled but not yet finished processing. If you crash after the auto-commit but before processing, those records are skipped on resume, which is message loss.
package com.bank.kafka;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.util.Properties;
public class AutoCommitDanger {
// WRONG: auto-commit risks message loss
public static Properties wrongConfig() {
Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"); // Default
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");
// Timer commits offsets for polled but unprocessed records
// Crash after auto-commit but before processing = LOSS
System.out.println("WRONG: auto-commit can lose unprocessed records");
return props;
}
// CORRECT: manual commit after processing
public static Properties correctConfig() {
Properties props = new Properties();
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
// Commit only after processing succeeds
// Crash before commit = safe reprocess (at-least-once)
System.out.println("CORRECT: manual commit after processing -- at-least-once");
return props;
}
public static void main(String[] args) {
wrongConfig();
correctConfig();
}
}
Expected output:
WRONG: auto-commit can lose unprocessed records
CORRECT: manual commit after processing -- at-least-once
How do you implement reliable at-least-once delivery?
Turn off auto-commit and commit manually after processing succeeds. Poll a batch, process each record fully including any database write, and only then call commitSync().
package com.bank.kafka;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class AtLeastOnceConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "settlement-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
consumer.subscribe(Collections.singletonList("transfers"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> record : records) {
// Step 1: Process the record (database write)
settleTransfer(record.key(), record.value());
}
if (!records.isEmpty()) {
// Step 2: Commit ONLY after all records are processed
consumer.commitSync();
System.out.println("Committed offsets for " + records.count() + " records");
}
}
}
}
private static void settleTransfer(String accountId, String payload) {
System.out.println("Settled: " + accountId + " -> " + payload);
}
}
Expected output:
Settled: SAV2001 -> DEBIT 50000
Settled: SAV3001 -> CREDIT 50000
Committed offsets for 2 records
Why is idempotent processing essential?
Since duplicates are possible with at-least-once, you design the downstream so applying the same record twice has the same effect as once. In banking you use an idempotency key, often the transfer id, and a processed-ids table with a unique constraint.
package com.bank.kafka;
import java.math.BigDecimal;
import java.util.HashSet;
import java.util.Set;
public class IdempotentSettlement {
// Simulates a processed-transfers table with unique constraint
private static final Set<String> processedTransfers = new HashSet<>();
public static boolean settle(String transferId, String accountId, BigDecimal amount) {
if (processedTransfers.contains(transferId)) {
System.out.println("SKIP: Transfer " + transferId + " already processed");
return false; // Idempotent -- no double charge
}
// Apply the settlement
processedTransfers.add(transferId);
System.out.println("SETTLED: " + transferId + " account=" + accountId
+ " amount=" + amount);
return true;
}
public static void main(String[] args) {
// First processing -- succeeds
settle("TXF-123", "SAV2001", new BigDecimal("50000"));
// Duplicate after crash and reprocess -- safely skipped
settle("TXF-123", "SAV2001", new BigDecimal("50000"));
// Different transfer -- succeeds
settle("TXF-124", "SAV3001", new BigDecimal("25000"));
}
}
Expected output:
SETTLED: TXF-123 account=SAV2001 amount=50000
SKIP: Transfer TXF-123 already processed
SETTLED: TXF-124 account=SAV3001 amount=25000
What is the difference between commitSync and commitAsync?
commitSync() blocks until the commit is acknowledged, giving certainty but adding latency. commitAsync() returns immediately with a callback for lower latency but can silently fail. A common pattern is commitAsync during normal processing and a final commitSync on shutdown or rebalance.
| Method | Behavior | Use Case |
|---|---|---|
| commitSync() | Blocks until commit acknowledged | Shutdown, rebalance listener |
| commitAsync() | Returns immediately, callback on result | Normal processing loop |
package com.bank.kafka;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;
public class CommitPatternDemo {
public static void demonstrateCommitPattern(KafkaConsumer<String, String> consumer) {
// Normal loop: async for throughput
consumer.commitAsync((offsets, exception) -> {
if (exception != null) {
System.out.println("Async commit failed: " + exception.getMessage());
} else {
System.out.println("Async commit succeeded");
}
});
// On shutdown or rebalance: sync for certainty
// consumer.commitSync();
}
}
What is the correct ordering of process, persist, and commit?
The ordering matters for correctness. Committing before persisting is at-most-once and can lose data. Persisting then committing is at-least-once and safe when idempotent.
| Order | Semantics | Risk |
|---|---|---|
| Commit, then process | At-most-once | Crash after commit loses records |
| Process, then commit | At-least-once | Crash before commit reprocesses |
| Process + commit atomically | Exactly-once | Requires Kafka transactions |
What are common mistakes with offset commit handling?
The most common mistake is persisting to the database and committing the offset in two separate, non-atomic operations. If the commit succeeds but the database write fails, the record is skipped. If the database write succeeds but the commit fails, the record is reprocessed. The fix is to make the consumer idempotent so reprocessing is safe, or to store the offset in the same database transaction as the business data.
package com.bank.kafka;
public class OffsetCommitMistakes {
public static void main(String[] args) {
System.out.println("WRONG: commit before persist");
System.out.println(" consumer.commitSync();");
System.out.println(" database.insert(record); // Fails -> record lost!");
System.out.println();
System.out.println("WRONG: persist then commit (without idempotency)");
System.out.println(" database.insert(record);");
System.out.println(" consumer.commitSync(); // Fails -> record reprocessed");
System.out.println(" // If insert is not idempotent -> duplicate row!");
System.out.println();
System.out.println("CORRECT: persist with idempotency key, then commit");
System.out.println(" database.upsert(record, idempotencyKey);");
System.out.println(" consumer.commitSync();");
System.out.println(" // Reprocess is safe: upsert is a no-op on duplicate");
}
}
Expected output:
WRONG: commit before persist
consumer.commitSync();
database.insert(record); // Fails -> record lost!
WRONG: persist then commit (without idempotency)
database.insert(record);
consumer.commitSync(); // Fails -> record reprocessed
// If insert is not idempotent -> duplicate row!
CORRECT: persist with idempotency key, then commit
database.upsert(record, idempotencyKey);
consumer.commitSync();
// Reprocess is safe: upsert is a no-op on duplicate
What should a senior engineer know?
For true exactly-once you need Kafka transactions that commit the output and the offset atomically, or an idempotent sink. Commit granularity matters: committing per record is slow, so you usually commit per batch, accepting that a crash reprocesses the whole batch. Handle commitAsync failures and do a final commitSync on shutdown via a ConsumerRebalanceListener.
Visual Overview
At-Least-Once Delivery Flow
============================
┌──────────────────┐
│ consumer.poll() │
│ offsets 50-59 │
└────────┬─────────┘
│
▼
┌──────────────────┐
│ Process records │
│ 50, 51, 52... │
└────────┬─────────┘
│
┌────┴────────────────┐
│ │
▼ ▼
┌──────────┐ ┌────────────────┐
│ SUCCESS │ │ CRASH at 55 │
└────┬─────┘ │ before commit │
│ └───────┬────────┘
▼ │
┌──────────┐ ▼
│ commitSync│ ┌────────────────┐
│ offset=60 │ │ Committed │
└──────────┘ │ offset = 50 │
│ (unchanged) │
└───────┬────────┘
│
▼
┌────────────────┐
│ Resume from 50 │
│ Reprocess │
│ 50-59 (dupes) │
└────────────────┘
Key: process THEN commit = at-least-once
Idempotency key stops double-charge
Key Takeaway
Kafka records a consumer group's position as a committed offset, and a record is only considered done once its offset is committed. If a consumer polls a batch, starts processing, and crashes before committing, the committed offset still points before that batch, and on resume the batch is reprocessed. Nothing is lost, but some records are processed more than once, which is at-least-once delivery.
The common mistake is relying on auto-commit, which commits on a timer and can advance past records you have not finished processing. For reliable at-least-once I disable auto-commit and commit manually only after processing succeeds. Since duplicates are now possible, I make processing idempotent using a transfer id with a processed-transfers table.
In my National Bank of India settlement service, we disabled auto-commit and committed offsets only after applying each transfer to the ledger, keyed by transfer id so reprocessing after a crash never double-charged Rajesh.
Remember one sentence: disable auto-commit, commit only after processing succeeds, and make the downstream idempotent so at-least-once delivery never causes a double charge.
---