Kafka consumer crashed before committing offsets and replayed stock movements twice

The stock count for about 140 SKUs in one warehouse went negative on a Tuesday morning. Pickers were being told to pick items the system said were not there, and the reconciliation job flagged the warehouse as "drifted". The drift appeared within one minute of a restart of the stock-ledger consumer, which had been killed by Kubernetes:

State:          Running
Last State:     Terminated
  Reason:       OOMKilled
  Exit Code:    137

Reproducing it

stock-ledger reads warehouse.movements (24 partitions, keyed by SKU) and appends one row per movement to a stock_movements table in PostgreSQL. Current stock is the sum of movements. The consumer used enable.auto.commit=true with the default auto.commit.interval.ms of 5000.

In staging we produced 50,000 movements at a steady rate and ran kill -9 on the consumer JVM. After restart, the table had between a few hundred and a few thousand extra rows, depending on when in the commit interval the kill landed.

What we ruled out

• Duplicate events from the producer. The producer runs with idempotence on, and the duplicated rows referenced identical Kafka coordinates, which we had started logging in a source column.
• A rebalance storm. The group did rebalance once, as expected when a member disappears, but there was no repeated join/leave loop. (Slow-poll rebalances are a separate failure we have written up before; this was not one.)
• A bug in the stock sum. Recomputing from distinct (partition, offset) pairs gave the correct numbers.

Root cause

With auto-commit, the Java consumer commits offsets of records returned by earlier poll() calls, on a timer, from inside the next poll(). Up to five seconds of processed records had their rows written to Postgres but their offsets not yet committed to Kafka. When the JVM died, the new owner of those partitions resumed from the older committed offset and inserted those movements again. The handler was a plain INSERT, so every replay counted.

The first "fix" a teammate shipped as a hotfix made it worse in a quieter way: committing synchronously right after poll(), before processing. That turns duplicates into gaps, because anything polled but not yet written is skipped after a crash. Our nightly reconciliation caught missing movements two days later.

What is happening under the hood

Kafka stores one number per group and partition: the committed offset, meaning "next record to read". Your application has a second piece of state: what it has actually applied. A crash between those two writes is unavoidable when they live in different systems:

Auto-commit is "process, then commit later", with a time window instead of a per-record one. Manual commitSync() after each batch narrows the window but does not remove it.

Kafka's transactional API gives atomic "consume, produce, commit" when the output is another Kafka topic. Our output is a PostgreSQL table, so the atomic boundary has to be the database transaction. The consumer API supports this directly: you can keep offsets anywhere and call seek() to tell the consumer where to start.

Environment

Minimal reproduction

// Original loop (simplified)
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
try (var consumer = new KafkaConsumer<String, String>(props)) {
    consumer.subscribe(List.of("warehouse.movements"));
    while (true) {
        for (var rec : consumer.poll(Duration.ofMillis(500))) {
            jdbc.update("INSERT INTO stock_movements (sku, qty, source) VALUES (?, ?, ?)",
                rec.key(), parseQty(rec.value()), rec.partition() + ":" + rec.offset());
        }
    }
}

Kill the process with kill -9 a few seconds after a burst and restart it. Count rows grouped by source with HAVING count(*) > 1.

Who is affected

Every Kafka consumer whose output is outside Kafka and that is not idempotent. Language and client do not matter; librdkafka-based clients behave the same way with their own auto-commit. Graceful shutdown hides the problem because close() commits; only hard kills (OOM, node loss, SIGKILL after a grace period) show it.

The fix

We moved the source of truth for offsets into PostgreSQL and write it in the same transaction as the movement rows. On partition assignment, the consumer seeks to the offset stored in the database.

CREATE TABLE consumer_offsets (
  group_id    text    NOT NULL,
  topic       text    NOT NULL,
  partition   int     NOT NULL,
  next_offset bigint  NOT NULL,
  PRIMARY KEY (group_id, topic, partition)
);

public final class StockLedgerConsumer {
    private static final String GROUP = "stock-ledger";
    private final KafkaConsumer<String, String> consumer;
    private final DataSource ds;

    void run() throws SQLException {
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumer.subscribe(List.of("warehouse.movements"), new ConsumerRebalanceListener() {
            @Override public void onPartitionsRevoked(Collection<TopicPartition> parts) { }
            @Override public void onPartitionsAssigned(Collection<TopicPartition> parts) {
                for (TopicPartition tp : parts) {
                    long next = loadOffset(tp);       // SELECT next_offset ... or -1
                    if (next >= 0) consumer.seek(tp, next);
                }
            }
        });
        while (running) {
            ConsumerRecords<String, String> batch = consumer.poll(Duration.ofMillis(500));
            if (batch.isEmpty()) continue;
            try (Connection c = ds.getConnection()) {
                c.setAutoCommit(false);
                for (TopicPartition tp : batch.partitions()) {
                    List<ConsumerRecord<String, String>> recs = batch.records(tp);
                    insertMovements(c, recs);
                    long next = recs.get(recs.size() - 1).offset() + 1;
                    upsertOffset(c, tp, next);
                }
                c.commit();                           // results and offsets together
            }
            consumer.commitAsync();                   // only for lag dashboards
        }
    }

    private void upsertOffset(Connection c, TopicPartition tp, long next) throws SQLException {
        try (var ps = c.prepareStatement("""
            INSERT INTO consumer_offsets (group_id, topic, partition, next_offset)
            VALUES (?, ?, ?, ?)
            ON CONFLICT (group_id, topic, partition)
            DO UPDATE SET next_offset = EXCLUDED.next_offset
            WHERE consumer_offsets.next_offset < EXCLUDED.next_offset""")) {
            ps.setString(1, GROUP); ps.setString(2, tp.topic());
            ps.setInt(3, tp.partition()); ps.setLong(4, next);
            ps.executeUpdate();
        }
    }
}

Points that matter:

• If the JVM dies before c.commit(), neither rows nor offsets are stored; the next owner seeks to the old offset and reprocesses cleanly.
• If it dies after c.commit() but before commitAsync(), Kafka's committed offset is stale, but we ignore it because onPartitionsAssigned seeks to the database value.
• The WHERE ... < EXCLUDED.next_offset guard stops a zombie consumer (one that lost its partitions but is still finishing a batch) from moving the offset backwards. For stronger zombie fencing we also check, inside the transaction, that the partition is still in consumer.assignment() before committing.
• We still call commitAsync() so kafka-consumer-groups.sh and our lag exporter show sensible numbers. Those commits are advisory.

A unique constraint on stock_movements(source) went in as a second line of defence. With offsets stored atomically it should never fire; if it does, something is wrong with the seek logic and we want a loud failure instead of a silent duplicate.

Alternatives and trade-offs

• Idempotent writes keyed by event id (INSERT ... ON CONFLICT DO NOTHING). Simpler, works when each event maps to one row. It does not help with side effects that are not inserts, such as incrementing a counter.
• Manual commitSync() after each batch. Shrinks the duplicate window to one batch. Fine for analytics sinks, not for a ledger.
• Kafka transactions with isolation.level=read_committed. The right tool when the consumer writes results to another topic and a connector loads them into the database.
• Never commit first. If you need "at most once", say so explicitly in the design; do not get it by accident.

How to test the fix

The staging chaos script we now run before every release of the service:

#!/usr/bin/env bash
set -euo pipefail
./produce-movements.sh --count 50000 --rate 2000 &
for i in 1 2 3 4 5; do
  sleep $((RANDOM % 8 + 3))
  kubectl -n warehouse delete pod -l app=stock-ledger --grace-period=0 --force
done
wait
sleep 60
psql "$DATABASE_URL" -Atc "SELECT count(*), count(DISTINCT source) FROM stock_movements"
# expected: 50000|50000

Confirming in production

After rollout we compared the database offsets with Kafka's committed offsets and log end offsets. The database value should always be equal to or ahead of Kafka's committed value:

SELECT partition, next_offset FROM consumer_offsets
WHERE group_id = 'stock-ledger' ORDER BY partition;

kafka-consumer-groups.sh --bootstrap-server broker-1.example.com:9092 \
  --describe --group stock-ledger

The nightly reconciliation (sum of movements versus physical cycle counts) has shown zero drift caused by replays since the change, across about thirty pod restarts including two node failures.

Regression guard

• The chaos script runs in the staging pipeline; the job fails if count and count(DISTINCT source) differ or do not match the produced count.
• An alert fires if any consumer_offsets.next_offset is behind the Kafka committed offset for more than ten minutes, which would mean the seek path is broken.
• The unique constraint on source turns any future regression into an exception rather than bad stock.

Security implications

None beyond data integrity. Duplicated or lost stock movements are not an access-control issue, but negative stock and phantom stock both have financial impact and can mask shrinkage. The offsets table is in the same schema and permissions as the ledger; the consumer's database role needs INSERT and UPDATE on it and nothing more.

Prevention checklist

• Decide per consumer whether it is at-least-once, at-most-once or effectively-once, and write it in the README.
• Do not use auto-commit for consumers that write money, stock or any ledger.
• When output is a database, store offsets in that database in the same transaction and seek on assignment.
• Hard-kill consumers in staging; graceful shutdown tests prove nothing about commits.
• Add a unique constraint on the event's source coordinates or id as a backstop.

References

• Apache Kafka documentation, Consumer configs: https://kafka.apache.org/documentation/#consumerconfigs
• KafkaConsumer Javadoc, Storing offsets outside Kafka: https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html
• Apache Kafka documentation, Message delivery semantics: https://kafka.apache.org/documentation/#semantics
• PostgreSQL INSERT ON CONFLICT: https://www.postgresql.org/docs/17/sql-insert.html