The Trade-Offs of Adaptive Key Salting in Kafka

[Brindal P]

At 2:14 AM, nine of your ten Kafka consumers are idle. The tenth is pinned at 100% CPU, its lag climbing by the second, and the coordinator has started evicting it for missing max.poll.interval.ms. Every eviction triggers a rebalance. Every rebalance stops the other nine consumers cold for the duration. You didn't lose a broker. You didn't lose a data center. You partitioned by user_id.

I've operated production streaming systems where this exact failure shows up under load. Not because anyone wrote bad code, but because hash(user_id) % num_partitions encodes an assumption that almost never holds: that traffic distributes uniformly across keys. It doesn't. It follows a power law, and the gap between "textbook Kafka" and "Kafka at 99th-percentile traffic" lives entirely in that gap.

The Zipfian Lie​

Partitioning tutorials treat key selection as a hash-modulo exercise: pick a key with decent cardinality, feed it to murmur2, done. That holds right up until real users show up. In production workloads, it's routine for the top 1% of keys to generate 20%+ of total message volume: a single enterprise tenant, a viral account, a retry-storm producer hammering one order ID. Hash-modulo doesn't care. It sends every message for that key to the same partition, deterministically, forever, because that's the one guarantee it's actually designed to give you: per-key ordering.

The trap is that ordering and load-balancing are in direct tension, and naive partitioning quietly picks ordering every time, including for the one key where you didn't want that trade-off made for you.

What Actually Breaks​

Skew isn't just "one partition is bigger." It cascades through three distinct failure modes, and conflating them is why most incident postmortems stop at "we added partitions" and the problem comes back three weeks later.

1. Page cache contention becomes head-of-line blocking. Kafka leans on the OS page cache for reads; a hot partition's active segment usually stays resident, which is fine on its own. The damage is at the broker level: request-handler and network threads are shared across every partition it hosts. Once consumers on the hot partition fall behind and start reading segments that have aged out of cache, those fetches become real disk I/O and occupy handler threads far longer than a cache hit would, which stalls idle, unrelated partitions colocated on the same broker.

2. Single-consumer processing loops starve their own partitions. A poll() call returns a batch spanning every partition a consumer owns, capped by max.poll.records, not per partition. A hot partition dominating that batch means a synchronous processing loop burns its whole time budget on it before the next poll. Records from the consumer's other partitions sit untouched, not because Kafka deprioritized them, but because the application did.

3. Rebalances cascade. Miss max.poll.interval.ms on the hot partition and the coordinator evicts you. With the eager assignor (still the effective default on plenty of clusters), every consumer in the group stops fetching until reassignment completes. Confluent's own benchmark on a 10-instance Streams app measured 37,138 ms of pause time under eager rebalancing versus 3,522 ms under cooperative-sticky, a 10x difference on identical hardware. Worse: the remaining consumers inherit more partitions, possibly the hot one, and miss their own deadlines next. That's the storm: Stable → PreparingRebalance → CompletingRebalance → Stable, looping, lag climbing, while broker metrics look healthy because this is client-coordinator traffic, not a data-plane failure.

Three Ways Out​

Strategy	Ordering guarantee	Cardinality cost	Best for
Virtual key salting	Per-(key, salt) only; needs a downstream merge	Fixed, small (K sub-keys per hot key)	A handful of measured outlier keys
Compound domain sharding	Full, per compound key	Scales with tenant × entity cardinality	Unknown-cardinality keys at design time
Dynamic adaptive spillover routing	Full for cold keys; salted only for hot ones	Proportional to actual skew, not worst case	Traffic where which keys are hot shifts over time

Virtual key salting appends a deterministic suffix to a known-hot key, userId#0 through userId#7, spreading one entity's traffic across K partitions. It's a real, load-bearing pattern; Pinterest's MemQ storage layer salts writes for exactly this reason. The cost is real too: per-key ordering disappears unless a downstream consumer re-aggregates by stripping the salt, and that consumer now needs a larger state store to hold the K salted sub-streams until it can re-key them back into one. You're trading producer-side balance for consumer-side memory. You now own that merge contract permanently.

Compound domain sharding widens the key itself, tenantId|entityId instead of tenantId alone, so you're not relying on one field's cardinality to carry the whole distribution. It's the right default when you don't yet know your traffic shape, but it won't save you from one dominant tenant; it just raises the bar for how bad the skew has to get before it's your problem.

Dynamic adaptive spillover routing is the pattern I reach for once hand-maintained salting becomes a maintenance burden: track key frequency at the producer, and only salt the keys that cross a measured hot-key threshold. The other 99% of keys, the ones that were never the problem, keep full ordering and pay zero added cost. This beats blanket salting for any traffic where the hot set shifts week to week, which, in practice, is most traffic.

Before vs. After: The Go Implementation​

The naive partitioner below is what most teams ship first. It's correct-looking, it compiles, and it has a bug that only surfaces under real hash distributions.

package partition

import (
	"hash/fnv"

	"github.com/IBM/sarama"
)

// NaiveKeyPartitioner is the textbook hash-modulo partitioner.
// No hot-key defense, and a sign bug: casting a uint32 hash to
// int32 flips the high bit on roughly half of all hash values,
// producing a negative dividend. Negative % positive in Go keeps
// the dividend's sign, so this can hand back a negative or
// out-of-range partition index under real traffic.
type NaiveKeyPartitioner struct{}

func NewNaiveKeyPartitioner(topic string) sarama.Partitioner {
	return &NaiveKeyPartitioner{}
}

func (p *NaiveKeyPartitioner) Partition(msg *sarama.ProducerMessage, numPartitions int32) (int32, error) {
	keyBytes, err := msg.Key.Encode()
	if err != nil {
		return 0, err
	}
	h := fnv.New32a()
	h.Write(keyBytes)
	return int32(h.Sum32()) % numPartitions, nil // bug: signed cast before modulo
}

func (p *NaiveKeyPartitioner) RequiresConsistency() bool { return true }

One caveat before going further: both code samples below hash with FNV-1a, matching sarama's own built-in HashPartitioner, not murmur2, which is what Kafka's Java client hashes with by default. That's irrelevant if your producers are all Go. It matters if Go and Java producers write the same keys to the same topic expecting them to land on the same partition: the two hash families won't agree, and nothing will warn you. Pick one hash function and enforce it across every producer language in the fleet, or don't depend on cross-language partition agreement at all.

The adaptive version fixes the sign bug, adds a lightweight hot-key sketch, and reuses buffers through sync.Pool instead of allocating a fresh one per call, since salting means building a new key on every hot message:

package partition

import (
	"bytes"
	"strconv"
	"sync"
	"sync/atomic"
	"time"

	"github.com/IBM/sarama"
)

const (
	trackerWidth  = 4096            // buckets in the heavy-hitter sketch
	hotThreshold  = 500             // msgs/window that mark a bucket "hot"
	saltFactor    = 8               // sub-partitions per hot key
	decayInterval = 2 * time.Second // sliding window length
)

// hotKeyTracker is a single-row, decaying frequency sketch: an
// approximation, not a full Count-Min Sketch. Each tick halves
// every counter via compare-and-swap instead of zeroing it: a
// hard reset would let a key sitting at 499 hits drop straight to
// 0 the instant the window rolls over, blind to bursts that
// straddle the boundary. Halving lets recent history bleed off
// gradually instead. For tighter false-positive control at real
// scale, swap this for a proper Count-Min Sketch or a
// Space-Saving top-K counter.
//
// Also per-process: see the note below the code for why that
// matters and how to fix it.
type hotKeyTracker struct {
	counters [trackerWidth]int64
}

func newHotKeyTracker() *hotKeyTracker {
	t := &hotKeyTracker{}
	go t.decayLoop()
	return t
}

func (t *hotKeyTracker) decayLoop() {
	ticker := time.NewTicker(decayInterval)
	defer ticker.Stop()
	for range ticker.C {
		for i := range t.counters {
			for {
				old := atomic.LoadInt64(&t.counters[i])
				if old == 0 {
					break
				}
				if atomic.CompareAndSwapInt64(&t.counters[i], old, old/2) {
					break
				}
			}
		}
	}
}

func (t *hotKeyTracker) touch(keyHash uint32) bool {
	b := int(keyHash % trackerWidth)
	n := atomic.AddInt64(&t.counters[b], 1)
	return n > hotThreshold
}

// AdaptiveSaltingPartitioner salts only the keys currently measured
// as hot. Cold keys (the overwhelming majority) keep strict
// per-key ordering and pay zero extra cost.
type AdaptiveSaltingPartitioner struct {
	tracker *hotKeyTracker
	saltSeq uint64
	bufPool sync.Pool
}

func NewAdaptiveSaltingPartitioner(topic string) sarama.Partitioner {
	return &AdaptiveSaltingPartitioner{
		tracker: newHotKeyTracker(),
		bufPool: sync.Pool{New: func() any { return new(bytes.Buffer) }},
	}
}

// fnv32a computes the FNV-1a hash of data directly, without going
// through hash/fnv's hash.Hash32 interface. Interface method
// dispatch on that type typically forces its state onto the heap
// on every call, hot key or not; a plain function operating on a
// value doesn't.
func fnv32a(data []byte) uint32 {
	const (
		offset32 = 2166136261
		prime32  = 16777619
	)
	h := uint32(offset32)
	for _, b := range data {
		h ^= uint32(b)
		h *= prime32
	}
	return h
}

func (p *AdaptiveSaltingPartitioner) Partition(msg *sarama.ProducerMessage, numPartitions int32) (int32, error) {
	keyBytes, err := msg.Key.Encode()
	if err != nil {
		return 0, err
	}
	keyHash := fnv32a(keyBytes)

	if !p.tracker.touch(keyHash) {
		return int32(keyHash % uint32(numPartitions)), nil // cold path: unsalted, fully ordered
	}

	buf := p.bufPool.Get().(*bytes.Buffer)
	buf.Reset()
	salt := atomic.AddUint64(&p.saltSeq, 1) % saltFactor
	buf.Write(keyBytes)
	buf.WriteByte('#')
	buf.WriteString(strconv.FormatUint(salt, 10))

	// The salted bytes have to become the message's actual key --
	// otherwise the partition returned here doesn't match what a
	// consumer sees on the wire, and there's nothing for a
	// downstream merge step to strip. sarama's Partitioner takes a
	// pointer to the full message, so rewriting msg.Key in place
	// is correct here -- no interceptor needed, unlike Java
	// clients, where the partitioner can't mutate the record.
	//
	// The copy out of buf is not optional: msg.Key has to outlive
	// this call, buf doesn't. It goes back in the pool right after,
	// and the next hot message can overwrite its backing array
	// before this one is even encoded onto the wire.
	salted := make([]byte, buf.Len())
	copy(salted, buf.Bytes())
	p.bufPool.Put(buf)

	msg.Key = sarama.ByteEncoder(salted)
	return int32(fnv32a(salted) % uint32(numPartitions)), nil
}

func (p *AdaptiveSaltingPartitioner) RequiresConsistency() bool { return true }

Wire it in with config.Producer.Partitioner = partition.NewAdaptiveSaltingPartitioner. Any downstream consumer doing per-key aggregation needs to know saltFactor and strip everything from # onward before grouping. That contract has to be documented and versioned, or your fix becomes next quarter's silent correctness bug.

Two things worth being upfront about. First, this tracker runs per producer process: a hot key spread evenly across a fleet of producers never crosses hotThreshold on any single instance, and slips past detection entirely. That needs coordination (a shared counter, or a centrally computed hot-key list pushed to producers), not a bigger local threshold. Second, don't take "allocation-conscious" on faith. Benchmark it:

// partition_test.go, same package, plus "testing" in the imports.
func BenchmarkAdaptiveSaltingPartitioner(b *testing.B) {
	p := NewAdaptiveSaltingPartitioner("bench-topic")
	msg := &sarama.ProducerMessage{}
	b.ReportAllocs()
	for i := 0; i < b.N; i++ {
		msg.Key = sarama.StringEncoder("hot-user-42")
		_, _ = p.Partition(msg, 12)
	}
}

Run that yourself before citing a number. I don't have a Go toolchain here to produce one. What I can say without running it: the pool removes the allocation a naive fmt.Sprintf would cost on every hot message, and fnv32a removes the one hash.Hash32 was quietly adding on every message, hot or cold. What's left is one allocation per hot message, copying the salted bytes into memory the message actually owns. That's not optional, for the reason above. Zero allocations on the hot path isn't achievable here without giving up that safety; near-zero is.

The Rebalance Isn't Free Either​

Salting fixes produce-side skew, but not what happens when a consumer falls behind for unrelated reasons: a slow downstream dependency, a bad deploy. That's a consumer-group problem, and it has its own fix: get off the eager assignor. CooperativeStickyAssignor (Kafka 2.4+) turns full-group stop-the-world pauses into incremental, partition-scoped handoffs. KIP-848, the next-generation consumer rebalance protocol, reached GA with Kafka 4.0 and removes eager rebalancing entirely, but it's opt-in on both sides. Clients need group.protocol=consumer; brokers need consumer included in group.coordinator.rebalance.protocols. Check both before assuming you already have it. A client flag with no matching broker config does nothing.

Checklist Before You Ship This​

• [ ] Profile before you provision: sample top-N keys by rate; top 1% of keys above ~20% of traffic means skew, not a capacity problem.
• [ ] Alert on lag skew, not raw lag: page on partition_lag_max / partition_lag_avg > 1.5.
• [ ] Don't reach for num.partitions first: a hot key still lands on exactly one partition no matter how many you add.
• [ ] Default to compound keys under uncertainty: tenantId|entityId costs nothing at low scale and saves a migration later.
• [ ] Salt measured outliers only: blanket salting breaks ordering nobody asked you to break.
• [ ] Version the merge contract: salt factor and strip logic are a public API to your consumers now.
• [ ] Move to cooperative-sticky or KIP-848: bound the blast radius of the rebalance that skew will eventually trigger.
• [ ] Rehearse the migration, not just the steady state: old data on a hot partition doesn't retroactively rebalance; test reassignment under production-like load before you need it at 2 AM.

Partitioning by user_id isn't wrong because it's simple. It's wrong because it silently bets your production stability on a uniform distribution your own traffic has never once produced.