Messages processed out of order across partitions

What this error actually means

Kafka guarantees order within a partition and nothing at all across partitions. Two events for the same entity that land on different partitions are consumed by different consumers, at different speeds, with no coordination between them. Whichever consumer happens to be less busy wins the race, and the last write to the database is not necessarily the newest event.

The result is a corrupted current state with a perfectly intact event log — which is why this is so hard to spot. Replaying the log in order produces the right answer, so nobody believes the pipeline is broken. Only the materialised view, search index, or cache is wrong, and only for the entities that happened to have two rapid updates.

The usual reason two events for one entity split across partitions is the partition key. No key at all means round-robin, so consecutive events for the same user land anywhere. A key chosen at the wrong granularity — `region` when you need `user_id`, or `user_id` when a transfer touches two accounts — has the same effect. And changing a topic’s partition count rehashes every key, so events published before the change sit on a different partition from events published after.

Note that per-partition order is necessary but not sufficient. A consumer that receives an ordered batch and then hands records to a thread pool has thrown the ordering away inside its own process, and this is a very common way to reintroduce the bug while believing the keys protect you.

How to diagnose it

kafka-console-consumer.sh --bootstrap-server broker:9092 --topic user-events \
  --from-beginning --property print.partition=true --property print.key=true \
  | grep '"user_id":1042'

SELECT id, last_event_ts, updated_at
FROM user_search_index
WHERE last_event_ts < (
  SELECT max(event_ts) FROM user_events e WHERE e.user_id = user_search_index.id
);

kafka-topics.sh --bootstrap-server broker:9092 --describe --topic user-events

The fix

Key the producer by the entity whose order you must preserve, so all events for that entity are serialised into one partition. This is the cheap structural fix and it covers the majority of cases — but it buys ordering at the price of a hot partition if one key is disproportionately busy, so check key distribution before committing to it.

Make every write reject stale data with a version guard, so ordering stops being load-bearing. Carry a monotonic version per entity — a database sequence, an LSN, an event sequence number — and apply the update only when the incoming version is greater than the stored one. This is the fix that survives partition changes, consumer concurrency, and multi-topic producers alike.

Do not use wall-clock timestamps as the version unless they come from a single clock. Producer clocks disagree by tens of milliseconds at best and minutes at worst, and two events within that window will be ordered arbitrarily. Where a timestamp is the only option, source it from the database on the write path rather than from the producer host.

If you must process concurrently for throughput, partition the concurrency by the same key you partitioned the topic with: hash the entity id to a fixed set of single-threaded workers. Order is preserved per entity while unrelated entities proceed in parallel.

Expand partition counts deliberately. Either accept a bounded window of disorder and rely on the version guard to absorb it, or drain the topic before the change, or create a new topic with the new partition count and cut consumers over. Adding partitions to a live keyed topic with no version guard reorders production data.

// Blind overwrite: the last event to be applied wins, whichever one that is.
jdbc.update("UPDATE user_index SET city = ? WHERE user_id = ?",
            evt.city(), evt.userId());

// Version guard: an older event cannot overwrite a newer one, regardless
// of which partition it arrived on or how many threads are consuming.
int applied = jdbc.update("""
        UPDATE user_index
           SET city = ?, version = ?, updated_at = now()
         WHERE user_id = ?
           AND version < ?
        """, evt.city(), evt.version(), evt.userId(), evt.version());
if (applied == 0) {
    metrics.counter("index.stale_event_dropped").increment();
}

-- for the insert-or-update case, keep the guard in the conflict clause
INSERT INTO user_index (user_id, city, version, updated_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (user_id) DO UPDATE
   SET city = EXCLUDED.city,
       version = EXCLUDED.version,
       updated_at = now()
 WHERE user_index.version < EXCLUDED.version;

How to stop it coming back

• Record the ordering key and the version field for every event type in a schema registry or contract test, so a new producer cannot omit them
• Count dropped stale events as a metric — a sudden rise means disorder increased, which is an early warning that a key or partition count changed
• Treat a partition-count increase as a migration with a written plan, not as a capacity dial
• Ban unkeyed produces to topics that feed materialised state; a lint rule or a producer wrapper is enough
• When adding concurrency to a consumer, make the key-affinity requirement explicit in the code rather than a comment — a plain executor is the default and it is wrong here

FAQ

Related

Other errors engineers hit next to this one

• Too many open files
• Out of memory: Killed process
• No space left on device despite free disk space
• Text file busy during executable replacement
• set -e script continues after a failed pipeline
• An unquoted variable turns one argument into several
• OOMKilled — container exit code 137
• CrashLoopBackOff