Kafka per-key ordering broke after we increased the partition count of a live topic

On a Monday afternoon our order-tracking projection started disagreeing with the payments ledger. Customers who had paid saw "Awaiting payment"; a few cancelled orders flipped back to "Paid". The projection service logged a burst of rejected transitions:

2025-03-03T13:41:22.118Z WARN  OrderProjector - rejected transition order=ord_51KQ from=CANCELLED to=PAID event=PaymentCaptured partition=14 offset=2210
2025-03-03T13:41:22.204Z WARN  OrderProjector - rejected transition order=ord_8VTA from=CREATED to=SHIPPED event=OrderShipped partition=9 offset=1873

The burst started at 13:38, two minutes after a platform engineer increased orders.status from 6 to 18 partitions to help a lagging consumer group.

Reproducing it

In staging: produce a continuous stream of status sequences (OrderCreated, PaymentCaptured, OrderShipped) keyed by order id, keep the consumer slightly behind, then run:

kafka-topics.sh --bootstrap-server broker-1.example.com:9092 \
  --alter --topic orders.status --partitions 18

Every run produced out-of-order transitions for orders that had events both before and after the change.

What we ruled out

• Producer retries reordering batches. Producers run with idempotence enabled (the default since Kafka 3.0), which preserves order per partition even with up to five in-flight requests.
• A consumer bug. The projector processes each partition sequentially, and the code had not changed in months.
• Clock skew. We order by Kafka offset, not event timestamps, and the rejected events had correct timestamps anyway.

Root cause

Kafka guarantees order only within a partition. The Java producer's default partitioner sends a keyed record to murmur2(key) mod partitionCount. When the count changed from 6 to 18, most keys started mapping to a different partition. Order ord_51KQ had OrderCancelled sitting unconsumed in partition 2 (old mapping) while its later PaymentCaptured retry went to the brand-new partition 14 (new mapping). Partition 14 had no backlog, so its consumer processed the newer event first. Two events for one key were now in two partitions, consumed by different threads at different speeds.

What is happening under the hood

The mapping is pure arithmetic, and you can compute it with the client library itself:

import org.apache.kafka.common.utils.Utils;
import java.nio.charset.StandardCharsets;

public class PartitionFor {
    public static void main(String[] args) {
        byte[] key = args[0].getBytes(StandardCharsets.UTF_8);
        for (int n : new int[] {6, 18}) {
            int p = Utils.toPositive(Utils.murmur2(key)) % n;
            System.out.printf("key=%s partitions=%d -> partition %d%n", args[0], n, p);
        }
    }
}

key=ord_51KQ partitions=6 -> partition 2
key=ord_51KQ partitions=18 -> partition 14

Going from N to 3N keeps a key in place only when its hash modulo 3N equals its hash modulo N, which is about one key in three. Two thirds of our order ids moved.

There is a second, quieter source of reordering. Producers learn the new partition count through metadata refresh, controlled by metadata.max.age.ms (five minutes by default). For a few minutes, a producer that refreshed early sends ord_X to the new partition while one that has not refreshed still sends it to the old one. Even with zero consumer lag, events for the same key can interleave across two partitions during that window.

Kafka cannot decrease partitions, so there is no undo.

Environment

Minimal reproduction

The projector's core, simplified:

@KafkaListener(topics = "orders.status", groupId = "order-projector")
public void on(OrderEvent e) {
    Order o = repo.find(e.orderId()).orElseGet(() -> Order.created(e.orderId()));
    if (!o.status().canMoveTo(e.targetStatus())) {
        log.warn("rejected transition order={} from={} to={}", e.orderId(), o.status(), e.targetStatus());
        return;
    }
    repo.save(o.withStatus(e.targetStatus()));
}

It trusts arrival order. That is correct while all of a key's events live in one partition, and wrong the moment they do not.

Who is affected

Any keyed topic whose consumers depend on per-key order: state machines, change-data-capture streams, account balances, inventory. Topics without keys (round-robin or sticky partitioning) do not lose any guarantee, because they never had per-key order. Kafka Streams applications are also affected: their state stores and changelog topics are co-partitioned with the input, and they refuse to start when partition counts of joined topics disagree.

The fix

We did three things: repaired the data, made the consumer tolerant of reordering, and wrote down a safe resize procedure.

1. Repair. We rebuilt the projection for affected orders from the ledger, which is the source of truth for payments and shipments. A script listed every order with a rejected transition after 13:36 and replayed its history from the database rather than from Kafka.

2. Version every event. Each producer now stamps a per-order sequence number taken from the order aggregate (the row's version column, incremented in the same transaction that emits the event). The projector applies an event only if it is the next version, ignores older ones, and parks newer ones:

public void on(OrderEvent e) {
    Order o = repo.findForUpdate(e.orderId()).orElseGet(() -> Order.empty(e.orderId()));
    if (e.version() <= o.version()) {
        return;                                   // duplicate or stale, safe to drop
    }
    if (e.version() > o.version() + 1) {
        parked.save(e);                           // gap: earlier event not seen yet
        return;
    }
    Order next = o.apply(e);
    repo.save(next);
    parked.drainInOrder(e.orderId(), next.version(), this::on);
}

A scheduled job alerts on parked events older than ten minutes, which would mean a real gap rather than a temporary reorder.

3. A resize procedure. For keyed topics we no longer run --alter --partitions on live traffic. We create a new topic and cut over:

# 1. create the target topic with the new count
kafka-topics.sh --bootstrap-server broker-1.example.com:9092 \
  --create --topic orders.status.v2 --partitions 18 --replication-factor 3

# 2. deploy consumers that read both topics (old first)
# 3. switch producers to orders.status.v2 (feature flag, all instances)
# 4. wait until the old topic has zero lag for the group
kafka-consumer-groups.sh --bootstrap-server broker-1.example.com:9092 \
  --describe --group order-projector | awk '$2=="orders.status" {sum+=$6} END {print sum}'
# 5. remove the old topic from consumers, then delete it after retention

Ordering holds because each key's old events are fully consumed before any of its new-topic events, as long as consumers do not start reading orders.status.v2 for a key before the old topic drains. For strictness we keep v2 consumers paused until step 4 reports zero, which costs a short processing pause instead of correctness.

If a pause is acceptable, the simpler variant works on the same topic: stop all producers, wait for zero lag, alter partitions, restart producers so they load fresh metadata, then resume.

Alternatives and trade-offs

• Custom partitioner with a fixed virtual partition count: hash to, say, 360 virtual slots and map slots to physical partitions. It reduces how many keys move, but any moved key still has the same problem, so you still need a drain.
• Over-provision at creation. Partitions are cheap compared with migrations. We now size keyed topics for expected traffic three years out.
• Rely only on versioning. Works for state machines with natural versions; not every event stream has one.

How to test the fix

An integration test using Testcontainers produces versioned events for the same order to two different partitions in reverse order and asserts the final state:

@Test
void outOfOrderEventsConvergeToLatestState() {
    send("orders.status", 1, ev("ord_T1", 1, CREATED));
    send("orders.status", 1, ev("ord_T1", 3, SHIPPED));   // newer, different partition
    send("orders.status", 0, ev("ord_T1", 2, PAID));       // older, arrives last
    await().atMost(Duration.ofSeconds(10))
        .until(() -> repo.find("ord_T1").orElseThrow().status() == SHIPPED);
    assertThat(parked.count("ord_T1")).isZero();
}

./mvnw -Dtest=OrderProjectorOrderingIT verify

Confirming in production

After the repair, a comparison between projection and ledger returned no mismatches:

SELECT p.order_id, p.status AS projected, l.status AS ledger
FROM order_projection p
JOIN ledger_order_status l USING (order_id)
WHERE p.status <> l.status;
-- expected: 0 rows

We then performed the v2 cutover for a different keyed topic using the procedure above, with the versioned projector live. Parked events peaked at 12 during the switch and all drained within seconds; no rejected transitions were logged.

Regression guard

• Topic changes go through our Terraform module; a policy check fails any plan that changes partitions on a topic tagged keyed=true.
• The rejected-transition log line is now a metric with an alert at more than five per minute.
• The ordering integration test runs in CI for the projector.

Security implications

Out-of-order state transitions briefly showed cancelled orders as paid, which could have let a customer receive goods for a cancelled payment if fulfilment had read the projection. Fulfilment reads the ledger, so no goods shipped incorrectly. The versioning change also hardens the consumer against replayed or duplicated events, which reduces the impact of anyone able to inject old events into the topic. Topic ACLs still restrict producers to the four services.

Prevention checklist

• Never alter the partition count of a keyed topic under live traffic.
• Size keyed topics generously at creation.
• Put a per-key version on events that drive state machines.
• Treat partition changes as a migration with a runbook, not a config tweak.
• Fix consumer lag by reducing per-record cost first; partitions are the last lever.

References

• Apache Kafka documentation, Modifying topics: https://kafka.apache.org/documentation/#basic_ops_modify_topic
• Apache Kafka documentation, Producer configs (metadata.max.age.ms, partitioner): https://kafka.apache.org/documentation/#producerconfigs
• Kafka source, BuiltInPartitioner: https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java
• Kafka Streams documentation, Partitioning requirements for joins: https://kafka.apache.org/documentation/streams/developer-guide/dsl-api.html