Implementing Custom Partitioners in Java and Python

The routing of a record from producer to partition determines ordering, data locality, and how evenly load spreads across your brokers. When the built-in partitioner does not align with your business logic, you can take control of that decision. This article covers how to implement a custom partitioner correctly in the Java client, and—because the Python client deliberately does not expose a custom-partitioner interface—how to achieve the same result from Python by computing the partition yourself. It includes verified APIs, the configuration that activates them, and the operational gotchas that bite in production.

How Kafka Partitions by Default

Before replacing the default behavior, be precise about what it actually does, because it has changed across releases:

  • With a key: the producer hashes the serialized key with murmur2 and takes the result modulo the number of partitions. This has been stable for years and is what guarantees that records with the same key land on the same partition (as long as the partition count does not change).
  • Without a key: the behavior is not simple round-robin. Since Kafka 2.4 (KIP-480), the producer uses sticky partitioning—it pins records to one partition until a batch fills, then switches. Since Kafka 3.3 (KIP-794), the built-in logic is the strictly uniform sticky partitioner, which assigns by bytes produced and steers away from slower brokers.

A further consequence of KIP-794: DefaultPartitioner and UniformStickyPartitioner are deprecated. You should not set partitioner.class to either of them. To get sticky behavior even for keyed records, leave partitioner.class unset and set partitioner.ignore.keys=true instead. A custom partitioner is for when you need routing logic that none of the built-in modes provide.

Why You Would Override Partitioning

The operational drivers for a custom partitioner are concrete:

Bounded fan-out for skewed tenants. Consider a fleet of IoT devices keyed by device_id. Hashing on device_id preserves per-device order, but if one enterprise customer owns 40% of the fleet, hashing can still pile a disproportionate share onto a few partitions. A custom partitioner can route on a composite key—spreading a heavy tenant across a bounded set of partitions while keeping per-device ordering within each.

Collocation on a logical sub-key. When your record key is a composite such as session_id:event_type but the grouping that matters for Kafka Streams local state is only session_id, a custom partitioner can extract and hash just that portion. This lets you collocate related events without forcing upstream producers to rewrite their keys—the same principle discussed in Keying Strategies for Ordered Message Processing.

Stable mapping across partition growth. The hard truth about partitioning is that hash(key) % N reshuffles almost every key when N changes. Adding partitions to a topic therefore breaks per-key ordering for in-flight data. A custom partitioner backed by a consistent-hashing ring or an explicit lookup table moves only a fraction of keys when partitions are added—an operational concern central to Topics, Partitions & Data Modeling. Note that this only helps records produced after the change; already-written records stay where they were.

The Java Partitioner Contract

A custom Java partitioner implements org.apache.kafka.clients.producer.Partitioner. The interface extends Configurable and Closeable, so you implement three methods:

  • configure(Map<String, ?> configs) — called once at startup with the producer configuration; use it to read your own custom properties.
  • int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) — called once per record. It must return a partition index in [0, numPartitions - 1].
  • close() — called at producer shutdown for resource cleanup.

The partition method receives the deserialized and serialized forms of both key and value plus the current Cluster metadata, from which you read cluster.partitionCountForTopic(topic) (returns Integer) or cluster.availablePartitionsForTopic(topic) if you want to avoid offline partitions. Returning an out-of-range index causes the send to fail, so validate.

Critically, partition runs synchronously on the calling thread inside the producer’s send path. It must be fast, deterministic, and side-effect-free. Any blocking call—a database lookup, a network round trip—throttles the entire send loop and shows up as producer latency. Precompute lookups and refresh them off the hot path.

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @Override
    public void configure(Map<String, ?> configs) {
        // Read custom configuration here
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        // Routing logic here
        return 0;
    }

    @Override
    public void close() {
        // Cleanup
    }
}

A Composite-Key Partitioner in Java

The most common production case: records carry a key of the form customerId:deviceId, and you want to partition by customerId so all of a customer’s devices collocate, while per-device order is preserved within the partition. The default partitioner would hash the whole composite string and scatter the devices. The custom partitioner parses out customerId and hashes only that.

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.utils.Utils;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;

public class CompositeKeyPartitioner implements Partitioner {

    private static final char DELIMITER = ':';

    @Override
    public void configure(Map<String, ?> configs) {
        // No custom configuration for this case
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        if (numPartitions <= 0) {
            throw new IllegalStateException("Topic " + topic + " has no partitions");
        }

        if (keyBytes == null || keyBytes.length == 0) {
            // No key: spread randomly. Do not derive a partition from the
            // thread id, which produces a fixed, skewed mapping.
            return ThreadLocalRandom.current().nextInt(numPartitions);
        }

        String keyStr = new String(keyBytes, StandardCharsets.UTF_8);
        int delimiterIndex = keyStr.indexOf(DELIMITER);
        String customerId = (delimiterIndex > 0)
                ? keyStr.substring(0, delimiterIndex)
                : keyStr;

        byte[] groupBytes = customerId.getBytes(StandardCharsets.UTF_8);
        return Utils.toPositive(Utils.murmur2(groupBytes)) % numPartitions;
    }

    @Override
    public void close() {
        // No resources to clean up
    }
}

Utils.murmur2(byte[]) and Utils.toPositive(int) are the same helpers Kafka’s own keyed partitioning uses, so this is consistent with how non-custom keyed records are placed. Two notes:

  • Using % numPartitions means this mapping changes if you add partitions—accept that, or replace the modulo with a lookup table or consistent hash if stable placement matters more than even spread.
  • The null-key branch above uses ThreadLocalRandom rather than the thread id. Deriving a partition from Thread.getId() maps a small, fixed set of threads to the same partitions for the life of the process, which is the skew you are trying to avoid.

Activate it by setting partitioner.class on the producer:

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("partitioner.class", "com.example.CompositeKeyPartitioner");

The standing operational risk is that the routing logic must stay stable across restarts and upgrades. If you change the delimiter or the hash, records with the same logical key move to different partitions and ordering breaks for that key. Roll changes out behind a canary and watch per-partition load (see Monitoring below) immediately afterward.

Custom Partitioning in Python

Here the draft most engineers start from is wrong, so be careful: the confluent-kafka Python client does not expose a custom-partitioner class. There is no confluent_kafka.Partitioner to subclass, no partition() callback, and no cluster object handed to your code. The feature has been requested repeatedly and declined upstream (it is not exposed from librdkafka into the Python binding). Any tutorial showing class MyPartitioner(confluent_kafka.Partitioner) is fabricated and will fail at import.

You have two correct options.

Option 1: Select a built-in partitioner by name

librdkafka, which backs confluent-kafka, ships several partitioners selected via the partitioner string config. The valid values are random, consistent, consistent_random, murmur2, murmur2_random, fnv1a, and fnv1a_random; the default is consistent_random. If you need keyed records to land on the same partition as a Java producer writing to the same topic, use murmur2_random, which matches the Java client’s hashing (with random placement for null/empty keys):

from confluent_kafka import Producer

producer = Producer({
    'bootstrap.servers': 'broker1:9092,broker2:9092',
    'partitioner': 'murmur2_random',  # Java-compatible key hashing
})

Option 2: Compute the partition in application code

For genuinely custom routing—composite keys, lookup tables, consistent hashing—compute the target partition yourself and pass it explicitly to produce() via the partition argument. When you supply partition, librdkafka uses it directly and skips the configured partitioner entirely.

import bisect
import hashlib
from confluent_kafka import Producer, KafkaException
from confluent_kafka.admin import AdminClient

class ConsistentHashRouter:
    """Maps keys to a stable partition using a hash ring with virtual nodes.
    Adding partitions moves only a fraction of keys instead of reshuffling all."""

    def __init__(self, num_partitions, virtual_nodes=4):
        self.virtual_nodes = virtual_nodes
        self._build_ring(num_partitions)

    @staticmethod
    def _hash(data: bytes) -> int:
        return int.from_bytes(hashlib.sha256(data).digest()[:8], 'big')

    def _build_ring(self, num_partitions):
        self.num_partitions = num_partitions
        ring = []
        for p in range(num_partitions):
            for v in range(self.virtual_nodes):
                point = self._hash(f"partition-{p}-vnode-{v}".encode())
                ring.append((point, p))
        ring.sort(key=lambda x: x[0])
        self._points = [point for point, _ in ring]
        self._owners = [owner for _, owner in ring]

    def partition_for(self, key_bytes: bytes) -> int:
        if not key_bytes:
            # Fall back to the broker's default by returning None upstream;
            # here we hash empty keys to a fixed point for determinism.
            key_bytes = b''
        h = self._hash(key_bytes)
        idx = bisect.bisect_right(self._points, h)
        if idx == len(self._points):
            idx = 0
        return self._owners[idx]


def get_partition_count(bootstrap, topic):
    md = AdminClient({'bootstrap.servers': bootstrap}).list_topics(topic, timeout=10)
    tmd = md.topics[topic]
    if tmd.error is not None:
        raise KafkaException(tmd.error)
    return len(tmd.partitions)


bootstrap = 'broker1:9092,broker2:9092'
topic = 'telemetry'
router = ConsistentHashRouter(get_partition_count(bootstrap, topic), virtual_nodes=4)
producer = Producer({'bootstrap.servers': bootstrap})

key = b'customer-42'
producer.produce(topic, key=key, value=b'...', partition=router.partition_for(key))
producer.flush()

The number of virtual nodes per partition controls evenness: more nodes smooth the distribution but enlarge the ring. Four to five per partition is a reasonable starting point. Because you fetch the partition count yourself (via AdminClient.list_topics, not a cluster object passed into a callback), refresh it deliberately—on a timer or on a partition-change event—and rebuild the ring when it changes. Pinning the count at process start means new partitions never receive traffic until you rebuild.

A caveat for Option 2: because you compute placement before handing the record off, your routing is only as fresh as your last partition-count refresh, and you bypass the broker-aware adaptive logic of the default partitioner. That trade is usually worth it when stable key placement is the goal.

Configuration and Deployment

The Java partitioner is loaded reflectively from partitioner.class, so it must be on the classpath of every producer JVM; a mismatch surfaces as ClassNotFoundException at startup. Bake the JAR into the container image rather than depending on a shared filesystem, publish it through a versioned artifact repository (internal Nexus/Artifactory or Maven Central), and pin the version in your build. Because partitioner.class takes a fully qualified class name, you can keep multiple versions side by side under different class names and cut over by config.

In Python there is no reflective lookup. With Option 1 the partitioner is a librdkafka string, so there is nothing to ship. With Option 2 the routing code is ordinary Python that must be importable by the producer process; package it in the image and ensure your CI builds it as a normal dependency.

Interaction with the idempotent producer

When enable.idempotence=true (the default since Kafka 3.0), the producer tags each record with a producer ID and a per-partition sequence number, and the broker uses those to deduplicate retries. The invariant the broker enforces is per (producer ID, partition). A custom partitioner is invoked once when the record is first assigned to a batch; retries of that batch go to the same partition, so idempotence is preserved as long as your partition method is deterministic for a given record. The failure mode to avoid is a partitioner whose output depends on mutable external state (a lookup table that changes mid-flight, a counter, wall-clock time): if the same logical record could be assigned different partitions across the producer’s internal handling, you undermine both ordering and the deduplication guarantee. Keep partition pure—input in, partition out—and idempotence and your custom routing coexist cleanly.

Note also that idempotence does not relax max.in.flight.requests.per.connection; the client caps it at 5 while preserving ordering. Your partitioner does not change that, but skew it creates will concentrate those in-flight requests onto a few partitions.

Monitoring for Skew

After deploying any partitioner change, confirm load is distributed as intended—do not assume. The producer-side JMX metrics live under kafka.producer:type=producer-topic-metrics,client-id=...,topic=... and give you per-topic record and byte rates, but Kafka does not publish a per-partition record rate on the producer. To see actual per-partition distribution, measure it broker-side:

  • Compare partition end offsets over a window with kafka-get-offsets.sh --bootstrap-server <b> --topic <t> --time -1 (the script is GetOffsetShell); growing the offset delta per partition is your throughput per partition.
  • Watch per-partition log size via the broker MBean kafka.log:type=Log,name=Size,topic=<t>,partition=<n>.

kafka-consumer-groups.sh reports consumer lag per partition, which is a useful secondary signal but reflects consumption, not the producer’s placement decisions—do not rely on it to validate a partitioner. Alert on a partition whose rate diverges sharply from the topic mean; that divergence is the first symptom of a hot partition introduced by a routing bug.

Summary

Custom partitioning is a precise tool, not a default reach. In Java, implement Partitioner, keep partition fast and deterministic, hash with Utils.murmur2/Utils.toPositive for Java-compatible placement, and set partitioner.class. In Python, accept that there is no custom-partitioner class: pick a built-in librdkafka partitioner by name (murmur2_random for Java compatibility), or compute the partition yourself and pass it to produce(partition=...). In both languages the same disciplines apply—determinism for idempotence, stable mapping across partition growth, and broker-side monitoring to catch skew before it becomes an incident.

Back to Keying Strategies For Ordered Message Processing