How is Data Distributed Across Partitions

Message distribution across topic partitions is handled on the client side by the producer's Partitioner component according to two fundamental scenarios:

🟢 Junior Level

30-Second Summary

Message distribution across topic partitions is handled on the client side by the producer’s Partitioner component according to two fundamental scenarios:

Message with a Key (key != null): Kafka uses deterministic hashing. The partitioner hashes the serialized key bytes using the MurmurHash2 algorithm and computes the modulo over the total partition count. All records sharing the same key (e.g., customerId = "cust-42") always route strictly to the same partition, preserving end-to-end FIFO processing order.
Message without a Key (key == null): Starting with Apache Kafka 2.4+, the producer uses Sticky Partitioning. It buffers records into batches assigned to one selected partition until the batch is full or linger.ms expires, then shifts to the next partition. This maximizes batch compression (LZ4/Zstandard), minimizes TCP packet overhead, and significantly boosts throughput.

Real-World Analogy

Think of partition distribution as package sorting at a regional postal logistics warehouse:

With a Key (Zip Code): Every package labeled with zip code 10001 (the key) is routed to Delivery Truck #1. Regardless of volume or arrival time, all packages for 10001 board the same truck and are delivered in the order they were processed.
Without a Key (Unsorted Bulk Mail): Packages without individual zip codes are general freight. Instead of tossing one envelope into Truck #1, the next into Truck #2, and the third into Truck #3 (legacy Round-Robin), the loader fills Truck #1 completely to the ceiling (Sticky Partitioning), shuts the cargo doors, and only then begins loading Truck #2.

Basic Java Producer Example

// 1. Keyed Message: Deterministic routing by key hash
ProducerRecord<String, String> keyedRecord = 
    new ProducerRecord<>("orders", "customer-42", "{\"item\":\"laptop\"}");
producer.send(keyedRecord); // Guaranteed to hit the exact same partition for customer-42

// 2. Unkeyed Message: Sticky Partitioning batching
ProducerRecord<String, String> unkeyedRecord = 
    new ProducerRecord<>("telemetry", null, "{\"temperature\": 21.5}");
producer.send(unkeyedRecord); // Bundled into the currently active partition batch

🟡 Middle Level

The Default Partitioner Algorithm & The Java Math Trap

Under the hood, Kafka’s standard DefaultPartitioner calculates the target partition index using:

\[\text{partition} = \text{Utils.toPositive}(\text{Utils.murmur2}(\text{keyBytes})) \pmod{\text{numPartitions}}\]

Why Not `Math.abs(key.hashCode()) % numPartitions`?

A notorious bug in custom partitioners stems from standard Java math:

// ❌ CRITICAL BUG in naive custom partitioners:
int partition = Math.abs(key.hashCode()) % numPartitions;

In Java, a signed 32-bit int spans from $-2^{31}$ (-2147483648) to $2^{31}-1$ (2147483647). Because two’s complement representation is asymmetric, Integer.MIN_VALUE has no positive counterpart:

Calling Math.abs(Integer.MIN_VALUE) evaluates to Integer.MIN_VALUE ($-2147483648$)!
The modulo operation Integer.MIN_VALUE % numPartitions yields a negative partition index (e.g., -2), immediately crashing the application with ArrayIndexOutOfBoundsException.

Kafka solves this with a bitwise mask in org.apache.kafka.common.utils.Utils:

public static int toPositive(int number) {
    return number & 0x7fffffff; // Clears the sign bit, forcing number >= 0
}

Evolution of Unkeyed Routing: Round-Robin vs. Sticky Partitioning

Metric / Aspect	Legacy Round-Robin (Kafka < 2.4)	Modern Sticky Partitioner (Kafka 2.4+)
Routing Pattern	Cycles record by record: P0 $\to$ P1 $\to$ P2 $\to$ P0	Fills a full batch on P0, then switches to P1
Network Packets	Many tiny, fragmented TCP packets	Fewer, large, dense TCP frames
Compression Ratio	Poor (single-record payloads fail dictionary compression)	Exceptional (LZ4/Zstd achieve high compression on large batches)
Producer & Broker Latency	High overhead, increased CPU context switches	Up to 50% p99 latency reduction under moderate load

Custom Partitioner Implementation

When business requirements demand non-uniform routing (e.g., VIP customers assigned to dedicated high-priority partitions while standard users share remaining partitions):

public class TieredCustomerPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (numPartitions < 2) {
            return 0;
        }

        // Dedicated partition 0 for enterprise tier
        if (key instanceof String stringKey && stringKey.startsWith("VIP_")) {
            return 0;
        }

        // Standard hash modulo distributed across partitions 1 .. (numPartitions - 1)
        int standardPartitionsCount = numPartitions - 1;
        int hash = Utils.toPositive(Utils.murmur2(keyBytes));
        return 1 + (hash % standardPartitionsCount);
    }

    @Override
    public void configure(Map<String, ?> configs) {}

    @Override
    public void close() {}
}

🔴 Senior Level

The “Hot Key” Skew Problem and Architectural Mitigations

In large-scale systems, key frequency typically follows a Pareto distribution ($80/20$ or $99/1$ rule):

If merchantId serves as the key, a global enterprise client publishing 50,000 events/sec saturates partition 3, causing massive consumer lag and disk saturation while other partitions sit idle.

Architectural Mitigation Patterns:

Compound Routing Keys: If chronological order is only required at the sub-entity level (e.g., per transaction or order rather than across the entire merchant):
\[\text{routingKey} = \text{merchantId} + \text{"\#"} + \text{orderId}\]
Transactions for the high-volume merchant scatter evenly across all partitions while maintaining strict sequential consistency for each unique order.

Bounded Key Salting: When high throughput must be distributed across a controlled set of partitions without breaking downstream processing:

// Spread the hot entity across a fixed pool of 4 virtual keys:
int salt = ThreadLocalRandom.current().nextInt(4);
String routedKey = "MEGA_MERCHANT_" + salt;

KIP-794: BuiltInPartitioner and Kafka 3.3+ Architecture

Starting in Apache Kafka 3.3.0, the legacy Partitioner interface and partitioner.class configuration were formally deprecated via KIP-794:

Routing logic moved directly into KafkaProducer and RecordAccumulator through BuiltInPartitioner.
Eliminates per-record heap allocations (garbage collection pressure) associated with calling external partitioner methods.
Introduces Adaptive Partitioning: the client tracks broker round-trip latency and queue depths. If a broker becomes slow or unresponsive, the partitioner automatically bypasses its partitions, routing traffic to healthy brokers and preventing cascading client buffer exhaustion.

Producer Batch Tuning: `linger.ms` and `batch.size` Edge Cases

batch.size (default 16 KB): Maximum memory footprint allocated per partition batch buffer.
linger.ms (default 0 ms): Artificial delay enabling records to coalesce into a single batch before network transmission.

The Degenerate Sticky Partitioning Trap: If linger.ms = 0 and traffic arrival rate is low (e.g., 1 event every 40 ms):

Each incoming message triggers an immediate socket flush because the sender thread does not wait.
Upon completing each flush, the sticky partitioner switches to the next partition.
Sticky partitioning degenerates into an inefficient round-robin pattern of 1-message batches.
Production Standard: Configure linger.ms = 5..20 ms for high-throughput producers to allow the sticky partitioner to bundle messages effectively.

4 Tricky Questions

1. Why does `Math.abs(key.hashCode()) % partitions.size()` periodically cause `ArrayIndexOutOfBoundsException` in production?

Answer: In Java, two’s complement integer arithmetic is asymmetric:

Integer.MIN_VALUE is $-2147483648$, whereas Integer.MAX_VALUE is $+2147483647$.
Attempting to negate Integer.MIN_VALUE causes 32-bit integer overflow; Math.abs(-2147483648) returns -2147483648 unchanged.
When computing (-2147483648) % N, Java preserves the negative sign, yielding a negative remainder (e.g., -3).
Attempting to access partitions.get(-3) throws ArrayIndexOutOfBoundsException.
Production Fix: Use bitwise masking key.hashCode() & 0x7fffffff or Kafka’s Utils.toPositive(hash).

2. Why does Kafka implement MurmurHash2 for partition key hashing instead of Java’s native `Object.hashCode()`?

Answer: For two primary reasons:

Polyglot Consistency: Kafka is a multi-language messaging backbone. Producers implemented in Go, Python, C++, Rust, and Java must route identical keys to identical partitions. Java’s String.hashCode() algorithm is JVM-specific. MurmurHash2 operates deterministically on raw serialized byte arrays (byte[]), ensuring cross-language interoperability.
Avalanche Effect and Uniform Distribution: Standard polynomial hashing (like String.hashCode()) suffers from collision clusters on sequential, short, or patterned strings. MurmurHash2 exhibits an exceptional avalanche property: changing a single bit in the key flips approximately 50% of the output hash bits, preventing artificial partition hotspots.

3. How does message routing behave when publishing records with a custom Java object key versus a raw `byte[]` or `String`?

Answer:

The Kafka Partitioner never calls .hashCode() or .equals() on the user’s Java domain object.
The ProducerRecord passes through the configured Serializer (e.g., Avro, Protobuf, JSON, StringSerializer) before the partitioner runs.
The partitioner operates exclusively on the resulting byte[] keyBytes.
Architectural Risk: If a serializer produces non-deterministic byte outputs (e.g., standard Java serialization with varying metadata, or JSON serializers with non-guaranteed field ordering), identical logical objects will yield different byte arrays, different MurmurHash2 values, and land on different partitions, completely destroying message ordering!

4. What happens when the broker hosting the leader of a sticky partition experiences high I/O latency or network degradation?