Ordering and Partitioning
Why message order only holds inside a partition, and how to pick a key so related events stay together and in sequence.
partition = hash(key) % numberOfPartitionsA Kafka topic is not one list. It is split into partitions, and a partition is a list of messages that only grows at the end. Partitions exist so one topic can be spread across several machines and read by several consumers at the same time.
Order holds inside a partition only
Within one partition, messages stay in the order they were appended, and a consumer reads them in that order. Across partitions there is no order at all. Two messages in different partitions can be seen in either sequence, because different consumers read them at different speeds. "The topic is ordered" is never true. "This partition is ordered" always is.
The key chooses the partition
A message can carry a key, a few bytes alongside the value. The producer hashes the key to a number and takes the remainder against the partition count. The same key always gives the same partition, so everything about that key stays in one ordered list. Pick as the key the thing whose history must stay in sequence: the order id, the account id, the device id. With no key the producer spreads messages freely and you have given up ordering.
What trips people up
The partition count is part of that remainder, so adding partitions sends existing keys somewhere new, and a key's old and new events end up in two partitions. Clients also hash differently: kafka-go's kafka.Hash uses FNV-1a, the Java client uses murmur2, so two producers on one topic can disagree unless you match them with kafka.Murmur2Balancer. A key that is too coarse, such as the country, leaves one partition overloaded and the rest idle. Finally, one partition goes to one group member, so the partition count caps how many consumers can work in parallel.
Example
package main
import (
"fmt"
"hash/fnv"
)
// partitionFor shows the rule a hashing producer follows: hash the key,
// then take the remainder against the number of partitions.
func partitionFor(key string, partitions uint32) uint32 {
h := fnv.New32a()
h.Write([]byte(key)) // this Write never returns an error
return h.Sum32() % partitions
}
func main() {
const partitions = 4
events := []struct {
key string
name string
}{
{"order-7", "created"},
{"order-8", "created"},
{"order-7", "paid"},
{"order-7", "shipped"},
}
// Every order-7 event lands together, so it is read in this sequence.
for _, e := range events {
fmt.Printf("%s %-7s -> partition %d\n", e.key, e.name, partitionFor(e.key, partitions))
}
}
// Output:
// order-7 created -> partition 3
// order-8 created -> partition 2
// order-7 paid -> partition 3
// order-7 shipped -> partition 3When to use it
- Key order events by order id so created, paid and shipped are applied in sequence.
- Diagnose a consumer that saw a payment event before the order it belongs to.
- Decide how many partitions a topic needs when you want eight workers reading it.
- Spot that a country-based key is sending most traffic to one partition.
More examples
Keying a producer so one order stays in sequence
This is the one-line fix for the most common ordering bug in Go services: the key was set, the Balancer was not, and a paid event overtook its created event. Reach for it whenever a consumer applies state changes per entity. If you also need ordering across producers written in other languages, agree on the hash first.
package main
import (
"context"
"log"
"github.com/segmentio/kafka-go"
)
func main() {
w := &kafka.Writer{
Addr: kafka.TCP("localhost:9092"),
Topic: "orders",
// kafka-go defaults to round-robin and ignores the key. Hash sends
// equal keys to one partition. Use Murmur2Balancer instead to match
// a Java producer writing to the same topic.
Balancer: &kafka.Hash{},
}
defer w.Close()
// Same key, so "created" is always read before "paid".
err := w.WriteMessages(context.Background(),
kafka.Message{Key: []byte("order-7"), Value: []byte(`{"event":"created"}`)},
kafka.Message{Key: []byte("order-7"), Value: []byte(`{"event":"paid"}`)},
)
if err != nil {
log.Fatal(err)
}
}Keeping per-key order inside your own service
Handling a partition with a pool of goroutines destroys the ordering the partition gave you. This pattern keeps it: hash the key again inside the process and give each key a fixed goroutine. Use it when one partition is too slow for a single handler but per-entity order still has to hold.
package main
import (
"fmt"
"hash/fnv"
"sync"
)
type event struct {
key string
name string
}
// Each worker owns one channel and one output slice, so no lock is needed.
func worker(in <-chan event, out *[]string, wg *sync.WaitGroup) {
defer wg.Done()
for e := range in {
*out = append(*out, e.key+" "+e.name)
}
}
func main() {
const n = 2
chans := make([]chan event, n)
handled := make([][]string, n)
var wg sync.WaitGroup
for i := range chans {
chans[i] = make(chan event, 4)
wg.Add(1)
go worker(chans[i], &handled[i], &wg)
}
h := fnv.New32a()
events := []event{{"order-7", "created"}, {"order-8", "created"}, {"order-7", "paid"}}
for _, e := range events {
h.Reset()
h.Write([]byte(e.key))
// Same key, same worker, so its events keep their order.
chans[h.Sum32()%n] <- e
}
for _, c := range chans {
close(c)
}
wg.Wait()
for i, got := range handled {
fmt.Println(i, got)
}
}
// Output:
// 0 [order-8 created]
// 1 [order-7 created order-7 paid]What adding a partition does to existing keys
Every key moved. Run this before you agree to grow a topic that other people depend on for ordering, because the events already stored stay where they are while new ones go elsewhere. The usual answers are to start with more partitions than you need, or to publish to a fresh topic and migrate consumers deliberately.
package main
import (
"fmt"
"hash/fnv"
)
func partitionFor(key string, partitions uint32) uint32 {
h := fnv.New32a()
h.Write([]byte(key))
return h.Sum32() % partitions
}
func main() {
// The same keys, before and after someone adds a fifth partition.
for _, key := range []string{"order-7", "order-8", "order-9"} {
fmt.Printf("%s: 4 partitions -> %d, 5 partitions -> %d\n",
key, partitionFor(key, 4), partitionFor(key, 5))
}
}
// Output:
// order-7: 4 partitions -> 3, 5 partitions -> 0
// order-8: 4 partitions -> 2, 5 partitions -> 3
// order-9: 4 partitions -> 1, 5 partitions -> 2
Discussion