Message Ordering and Partitioning: Where Ordering Guarantees Begin and End

Version note: The code in this post was built and run with C# 14 and .NET 10, against Apache Kafka 4.2 (a Docker container).

Guess First

Ada places an order, a few seconds later changes the delivery address from Izmir to Ankara, and a minute after that cancels the order. Ordering publishes three events; each carries the order's version at that moment:

using System.Text.Json.Serialization;

namespace Bookstore.Ordering.Contracts;

// Version: an increasing sequence number for one order's events (the aggregate's version)
[JsonPolymorphic(TypeDiscriminatorPropertyName = "type")]
[JsonDerivedType(typeof(OrderPlaced), "order-placed")]
[JsonDerivedType(typeof(ShippingAddressChanged), "shipping-address-changed")]
[JsonDerivedType(typeof(OrderCancelled), "order-cancelled")]
public abstract record OrderEvent(Guid OrderId, int Version);

public sealed record OrderPlaced(Guid OrderId, int Version, string Customer, string City) : OrderEvent(OrderId, Version);
public sealed record ShippingAddressChanged(Guid OrderId, int Version, string City) : OrderEvent(OrderId, Version);
public sealed record OrderCancelled(Guid OrderId, int Version) : OrderEvent(OrderId, Version);

Shipping builds a shipment plan from these events. Because of the campaign, Shipping's consumer has been scaled to two instances reading from the same queue. The instance that gets ShippingAddressChanged is a bit slow, the one that gets OrderCancelled is fast. The events are processed in Shipping in the order 1, 3, 2. What's the final state of the shipment plan? Does the order Ada cancelled get shipped?

Where Does Ordering Break?

Let's write two consumers first. One applies events in the order they arrive, the other looks at the version:

using Bookstore.Ordering.Contracts;

namespace Bookstore.Shipping;

public enum PlanStatus { ReadyToShip, Cancelled }

public sealed class ShipmentPlan(Guid orderId)
{
    public Guid OrderId { get; } = orderId;
    public PlanStatus Status { get; set; }
    public string City { get; set; } = "";
    public int LastVersion { get; set; }
    public override string ToString() => $"{Status}, {City}, v{LastVersion}";
}

// A consumer that trusts the order: applies events in whatever order they arrive
public sealed class NaiveShipmentPlanner
{
    public Dictionary<Guid, ShipmentPlan> Plans { get; } = [];

    public void Apply(OrderEvent e)
    {
        ShipmentPlan plan = Plans.TryGetValue(e.OrderId, out ShipmentPlan? p) ? p : Plans[e.OrderId] = new(e.OrderId);
        switch (e)
        {
            case OrderPlaced placed: plan.City = placed.City; plan.Status = PlanStatus.ReadyToShip; break;
            case ShippingAddressChanged moved: plan.City = moved.City; plan.Status = PlanStatus.ReadyToShip; break;
            case OrderCancelled: plan.Status = PlanStatus.Cancelled; break;
        }
        plan.LastVersion = e.Version;
    }
}

// A consumer that doesn't trust the order: rejects events with an old version
public sealed class VersionedShipmentPlanner
{
    public Dictionary<Guid, ShipmentPlan> Plans { get; } = [];
    public List<string> Rejected { get; } = [];

    public void Apply(OrderEvent e)
    {
        ShipmentPlan plan = Plans.TryGetValue(e.OrderId, out ShipmentPlan? p) ? p : Plans[e.OrderId] = new(e.OrderId);
        if (e.Version <= plan.LastVersion)
        {
            Rejected.Add($"{e.GetType().Name} v{e.Version} (already at v{plan.LastVersion})");
            return;
        }
        switch (e)
        {
            case OrderPlaced placed: plan.City = placed.City; break;
            case ShippingAddressChanged moved: plan.City = moved.City; break;
            case OrderCancelled: plan.Status = PlanStatus.Cancelled; break;
        }
        plan.LastVersion = e.Version;
    }
}

using System.Text.Json;
using Bookstore.Ordering.Contracts;
using Bookstore.Shipping;
using Confluent.Kafka;
using Confluent.Kafka.Admin;

// Three events of the same order, processed out of order because of two competing consumers
Guid adaOrder = Guid.CreateVersion7();
OrderEvent[] published =
[
    new OrderPlaced(adaOrder, 1, "Ada", "Izmir"),
    new ShippingAddressChanged(adaOrder, 2, "Ankara"),
    new OrderCancelled(adaOrder, 3),
];
OrderEvent[] processed = [published[0], published[2], published[1]];   // v2 was processed after v3

NaiveShipmentPlanner naive = new();
VersionedShipmentPlanner versioned = new();
foreach (OrderEvent e in processed) { naive.Apply(e); versioned.Apply(e); }
Console.WriteLine($"[naive]     {naive.Plans[adaOrder]}");
Console.WriteLine($"[versioned] {versioned.Plans[adaOrder]}; rejected: {string.Join(", ", versioned.Rejected)}");
Console.WriteLine();

[naive]     ReadyToShip, Ankara, v2
[versioned] Cancelled, Izmir, v3; rejected: ShippingAddressChanged v2 (already at v3)

The final state of the consumer that trusts the order is "ready to ship, Ankara". A cancelled order is going to be shipped, because the address change handler, arriving after the cancellation, pulled the plan back to "ready". The consumer that looks at the version rejected v2 with "I'm already at v3", and the order stayed cancelled.

No bug was needed for this mix-up; two natural behaviors from the previous two posts were enough. Competing consumers process messages in parallel, and which one finishes first isn't known. And with at-least-once delivery, when a message is redelivered, the messages behind it may already have been processed. Microsoft's Queue-Based Load Leveling guidance gives the same warning: when several consumers run in parallel, a queue doesn't preserve the submission order under all conditions. The book Enterprise Integration Patterns (EIP: Gregor Hohpe and Bobby Woolf, 2003) meant exactly this when it listed sequence issues among the challenges of messaging.

No Global Order, but Order per Key

If you want every message in the system processed in a single global order, you have to read from a single sequence with a single consumer: no parallelism, no scale. Fortunately, a global order is rarely needed. The events of Ada's order must be ordered relative to each other; the relative order of the events of Ada's order and Alan's order doesn't matter to anyone. The need is order per key: the same aggregate's events in sequence, different aggregates in parallel.

Brokers offer this under different names. In Kafka, it's the partition key: records with the same key go to the same partition, a partition is ordered internally, and within a consumer group a partition is read by one consumer at a time. In Azure Service Bus, it's sessions: messages with the same SessionId are delivered in FIFO (first in, first out) order, and at any time only to the single receiver holding that session's lock. In RabbitMQ, it's single active consumer: even with several consumers attached to a queue, only one is active and the others wait on standby. That preserves order but also cuts parallelism to one per queue; for parallelism, RabbitMQ's consistent hash exchange or super streams set up a separate queue/stream per key range.

Partition Keys in Kafka

Let's write Bookstore's Ordering events to a single topic, with the order's id as the key. For comparison, let's also write one order's events with the wrong key, each event's own id:

// ---------- Kafka: key = OrderId, one order's events in the same partition ----------
const string bootstrap = "localhost:9092";
const string topic = "eda05.orders";
using IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build();
await admin.CreateTopicsAsync([new TopicSpecification { Name = topic, NumPartitions = 3, ReplicationFactor = 1 }]);

ProducerConfig producerConfig = new()
{
    BootstrapServers = bootstrap,
    Partitioner = Partitioner.Murmur2Random,   // same hash as the Java producer
};
using IProducer<string, string> producer = new ProducerBuilder<string, string>(producerConfig).Build();

async Task<int> PublishAsync(string key, OrderEvent e, IProducer<string, string>? via = null)
{
    DeliveryResult<string, string> result = await (via ?? producer).ProduceAsync(topic,
        new Message<string, string> { Key = key, Value = JsonSerializer.Serialize(e) });
    return result.Partition.Value;
}

string[] customers = ["Ada", "Alan", "Grace"];
Dictionary<string, Guid> orderOf = customers.ToDictionary(c => c, _ => Guid.CreateVersion7());
foreach (string c in customers)
{
    Guid id = orderOf[c];
    OrderEvent[] events = [new OrderPlaced(id, 1, c, "Izmir"), new ShippingAddressChanged(id, 2, "Ankara"), new OrderCancelled(id, 3)];
    List<int> partitions = [];
    foreach (OrderEvent e in events) partitions.Add(await PublishAsync(key: id.ToString(), e));
    Console.WriteLine($"key=OrderId    {c,-6} v1..v3 -> partitions {string.Join(",", partitions)}");
}

// Wrong key: each event's own id
{
    Guid id = Guid.CreateVersion7();
    OrderEvent[] events = [new OrderPlaced(id, 1, "Linus", "Izmir"), new ShippingAddressChanged(id, 2, "Ankara"), new OrderCancelled(id, 3)];
    List<int> partitions = [];
    foreach (OrderEvent e in events) partitions.Add(await PublishAsync(key: Guid.NewGuid().ToString(), e));
    Console.WriteLine($"key=MessageId  Linus  v1..v3 -> partitions {string.Join(",", partitions)}");
}
Console.WriteLine();

key=OrderId    Ada    v1..v3 -> partitions 2,2,2
key=OrderId    Alan   v1..v3 -> partitions 2,2,2
key=OrderId    Grace  v1..v3 -> partitions 2,2,2
key=MessageId  Linus  v1..v3 -> partitions 2,0,2

When the key is OrderId, each order's three events went to the same partition; in this run the three orders happened to land in the same partition, but each is ordered internally. When the key is MessageId, v2 of Linus's order went to another partition. Now another consumer may process it, and nothing prevents it from being processed before v1 or after v3. The relationship between partitions, consumers, and the choice of key:

The Partitioner.Murmur2Random line in the code is there to guard against a trap. The default partitioner of librdkafka, which sits underneath Confluent.Kafka, uses a CRC32 (Cyclic Redundancy Check) hash of the key; the Java producer's default is murmur2. If both .NET and Java services write to the same topic and nothing is configured, the same order id can go to different partitions in the two languages, and order per key silently breaks. The librdkafka documentation describes murmur2_random as the functional equivalent of Java's default.

Ordering has a condition on the producer side too: retries. If a record fails with a transient error and is resent while the record behind it has already been written, the order can change. Kafka's idempotent producer (the sequence-numbered sending we saw in the previous post) closes this case too: with idempotence on, up to five concurrent requests per connection can be in flight without breaking order. On the .NET side that takes EnableIdempotence = true; in librdkafka it's off by default.

Key Choice and Hot Partitions

A key has two jobs: deciding order and spreading load. What happens if, during the campaign, most orders go to a single book and someone has said "let's order stock-related events by book" and made the key BookId?

// ---------- Hot partition: the choice of key decides how load is spread ----------
string[] books = ["Dune", "Emma", "The Hobbit", "Middlemarch"];
int[] byBook = new int[3], byOrder = new int[3];
for (int i = 0; i < 1000; i++)
{
    string book = i % 10 < 7 ? "Dune" : books[i % 4];   // ~70% of campaign orders are Dune
    Guid id = Guid.CreateVersion7();
    OrderPlaced e = new(id, 1, "Ada", "Izmir");
    byBook[await PublishAsync(key: book, e)]++;
    byOrder[await PublishAsync(key: id.ToString(), e)]++;
}
Console.WriteLine($"key=BookId   per partition: {string.Join(" / ", byBook)}");
Console.WriteLine($"key=OrderId  per partition: {string.Join(" / ", byOrder)}");
Console.WriteLine();

key=BookId   per partition: 850 / 100 / 50
key=OrderId  per partition: 335 / 343 / 322

The same thousand events, keyed by BookId, were spread across partitions as 850 / 100 / 50: all of Dune's events went to one partition. The single consumer reading that partition carries the entire campaign load while the others are almost idle. Adding consumers doesn't help either, because one partition can be read by one consumer at a time. Keyed by OrderId, the spread is 335 / 343 / 322; since orders are independent of each other, the load is balanced. This is called a hot partition, and its source is almost always a key with an uneven distribution.

Bookstore's decision: the key for Ordering events is OrderId. The ordering need lines up with the order aggregate's boundary, and orders are spread evenly. CustomerId may look attractive ("a customer's orders should be in order"), but a school ordering 400 books heats up its own partition. The rule from the aggregate posts in the Domain-Driven Design series helps here too: the consistency boundary is the aggregate, and the ordering need usually ends with the aggregate as well.

Changing the Partition Count

If the partition count is the upper limit of parallelism, adding partitions as load grows is the first solution that comes to mind. But the path from key to partition is a hash and a division: hash(key) % partitionCount. Change the divisor and the result changes:

// ---------- Adding partitions: the same key may move to another partition ----------
Guid[] openOrders = [.. Enumerable.Range(0, 12).Select(_ => Guid.CreateVersion7())];
Dictionary<Guid, int> before = [];
foreach (Guid id in openOrders) before[id] = await PublishAsync(id.ToString(), new OrderPlaced(id, 1, "Ada", "Izmir"));
await admin.CreatePartitionsAsync([new PartitionsSpecification { Topic = topic, IncreaseTo = 4 }]);
// A running producer refreshes metadata periodically (5 min by default); simulate a freshly deployed instance
using IProducer<string, string> redeployed = new ProducerBuilder<string, string>(producerConfig).Build();
List<string> moved = [];
foreach (Guid id in openOrders)
{
    int after = await PublishAsync(id.ToString(), new OrderCancelled(id, 2), redeployed);
    if (after != before[id]) moved.Add($"p{before[id]}->p{after}");
}
Console.WriteLine($"3 -> 4 partitions: {moved.Count} of {openOrders.Length} orders moved ({string.Join(", ", moved)})");

await admin.DeleteTopicsAsync([topic]);

3 -> 4 partitions: 11 of 12 orders moved (p0->p2, p1->p2, p0->p1, p2->p1, p2->p0, p2->p3, p2->p1, p2->p0, p0->p2, p1->p2, p2->p0)

For eleven of twelve open orders, the next event went to a different partition. While an order's old events still wait to be processed in the old partition, the new ones can be processed in the new partition by another consumer; during the transition there is no per-order ordering guarantee. On my first attempt I increased the partition count, waited three seconds, and sent with the same producer; no key moved, because the running producer hadn't yet learned the topic's new partition count (librdkafka refreshes metadata every five minutes by default). So during the transition, different producer instances can even write the same key to different partitions. The practical consequence: choose the partition count up front, a bit above the expected load, and treat increasing it later as a migration from an ordering point of view.

Not Depending on Order

Partition keys and sessions preserve order in the broker, but parallelism inside the consumer, retries, and reprocessing from the dead-letter queue can still break it. The sturdier path is designing the consumer so it doesn't need order. There are three techniques.

A version or sequence field. The VersionedShipmentPlanner above does this: each event carries the aggregate's version, and the consumer rejects a version older than what it knows. In the Domain-Driven Design series post on storing aggregates with Entity Framework (EF) Core, "Persisting Aggregates with EF Core 10", we added Order.Version as a concurrency token that increased with every change; put the same number into the event and the consumer has a ready-made sequence number. This approach works best when the event carries the relevant state (event-carried state transfer from post 02 of the series): when v3 arrives you don't need v2's content. If events carry only the difference, you need to notice the gap and either hold back until the missing event arrives or read the state from the source.

Commutative updates. Operations whose result doesn't change when their order changes don't need ordering. For stock, differences like "+2" and "-1" give the same total in whatever order they're applied. State transitions, on the other hand, aren't commutative: the order of "cancelled" and "address changed" changes the result. A caveat: a commutative difference isn't idempotent; apply "+2" twice and the stock is wrong. That's exactly the subject of the next post in the series.

Putting the state machine in the consumer. If Shipping enforces the rule "a cancelled plan can't go back to ready" on its own side, the cancellation wins whatever order the events arrive in. This defends not the order but the domain rule.

The Trade-off: Order, Parallelism, Throughput

Bookstore's choice is the second row plus the discipline of the third: Ordering events are partitioned by the OrderId key, and Shipping and Inventory also check the version in the events. The broker's ordering is an optimization; the consumer's check is the safety net.

Common Pitfalls

Not giving a key, or using the message id as the key

Records without a key are spread across partitions, and the same order's events can go to different partitions. Using each event's own id as the key gives the same result; that's what we saw with Linus's order. The key should be the identity of the thing whose order you want preserved.

Writing the same aggregate's events to different topics

If OrderPlaced is written to one topic and OrderCancelled to another, there is no ordering guarantee between them at all, even if they use the same key. Events whose order depends on each other should be in the same topic with the same key.

Parallelizing inside the consumer

Reading a partition in order and then handing each message to its own task with Task.Run breaks, inside the consumer, the order the broker preserved. If you need parallelism, build something that processes in order per key (such as several Channel<T> instances split by key), or increase the partition count.

Treating the ordering guarantee as the consumer's only defense

A session or partition key doesn't protect a message resent from the dead-letter queue hours later: that message arrives after all the events that came after it. A consumer without a version check reapplies an old state.

A Small Experiment

★☆☆ Change the processed array to [published[2], published[0], published[1]]: the cancellation arrives before the order. What do the two planners end up with? Is that result correct for the planner that looks at the version?

★★☆ Without adding a version to NaiveShipmentPlanner, add only a domain rule: a cancelled plan can't become ReadyToShip again through another event. Is the result correct in the opening scenario? In which scenario is it still wrong?

★★★ Remove the Partitioner line from the producer (the librdkafka default) and compare which partitions the same twelve order ids go to versus Murmur2Random. How many land in a different partition? How would you notice this in a system where a Java service writes to the same topic?

The version check rejected an old event; but what if the same event arrives twice? When v3 arrives a second time it's rejected, because its version isn't older, it's equal. But what if the event were a stock difference (+2) or a payment command? In the next post we'll look at idempotent consumers: what happens when the same message arrives twice, and why the dedup record has to be in the same transaction as the business data.