Implementing Custom Partitioners
1. Understanding Partitioner Concept
A Kafka partitioner is responsible for determining which partition a message should be sent to within a topic. It's a key component in Kafka's message routing mechanism, ensuring that messages are distributed across partitions according to a specific strategy.
| Property | Description | Default Value |
|---|---|---|
| partitioner.class | Class name of the partitioner to use. Can be a custom implementation. | org.apache.kafka.clients.producer.RoundRobinPartitioner |
| partitioner.availability.timeout.ms | Timeout for partition availability check. | 0 |
Example: Default Partitioner Behavior
Properties props = new Properties();
props.put("bootstrap.servers", "localhost: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", "org.apache.kafka.clients.producer.RoundRobinPartitioner");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));
Note: The default partitioner uses a round-robin strategy to distribute messages across partitions.
Key Points
- Partitioners ensure message distribution across topic partitions.
- Custom partitioners can implement specific routing logic.
- The default
RoundRobinPartitionerdistributes messages evenly.
2. Creating Custom Partitioner
| Property | Description | Example |
|---|---|---|
| partitioner.class | Class name of the custom partitioner to use | com.example.MyPartitioner |
| partitioner.availability.timeout.ms | Timeout for partition availability check | 1000 |
Example: Custom Partitioner Implementation
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
public class MyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
int numPartitions = cluster.partitionCountForTopic(topic);
if (keyBytes == null) {
return 0;
}
return Math.abs(keyBytes.hashCode()) % numPartitions;
}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) {}
}Note: The partitioner must implement the
partition method to determine which partition a message should go to.3. Implementing partition() Method
| Property | Description | Example |
|---|---|---|
| partition() | Method to determine which partition a message should be sent to | public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) |
Example: Custom Partitioner Implementation
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
public class CustomPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
int numPartitions = cluster.partitionCountForTopic(topic);
if (key == null) {
return 0;
}
return Math.abs(key.hashCode()) % numPartitions;
}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) {}
}Note: The
partition() method must be deterministic to ensure consistent message routing.4. Configuring Partitioner Class
| Property | Values | Description |
|---|---|---|
| partitioner.class | org.apache.kafka.clients.producer.RoundRobinPartitioner |
Class name of the partitioner to use. Default is org.apache.kafka.clients.producer.DefaultPartitioner. |
| partitioner.availability.timeout.ms | 0 |
Timeout for partition availability check in milliseconds. Used by RoundRobinPartitioner. |
Example: Custom Partitioner Configuration
Properties props = new Properties();
props.put("bootstrap.servers", "localhost: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.MyCustomPartitioner");
Note: The custom partitioner must implement
org.apache.kafka.clients.producer.Partitioner.
5. Using Hash-Based Partitioning
Hash-based partitioning is a common strategy for distributing messages across Kafka partitions. It uses a hash function on a key to determine the partition.
| Property | Value | Description |
|---|---|---|
| partitioner.class | org.apache.kafka.clients.producer.RoundRobinPartitioner |
Default partitioner class used if not specified |
| partitioner.class | org.apache.kafka.clients.producer.UniformStickyPartitioner |
Alternative partitioner that ensures sticky partitioning for a batch |
Example: Hash-Based Partitioning Configuration
Properties props = new Properties();
props.put("bootstrap.servers", "localhost: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", "org.apache.kafka.clients.producer.RoundRobinPartitioner");
Note: When using hash-based partitioning, the key must be provided to ensure consistent message routing to the same partition.
Example: Producer with Key for Hash Partitioning
ProducerRecord<String, String> record = new ProducerRecord<>(
"my-topic",
"user-123", // Key for hash-based partitioning
"message-content"
);
producer.send(record);
6. Implementing Round-Robin Partitioning
Round-robin partitioning is a simple and effective strategy for distributing messages across Kafka partitions in a balanced manner. It ensures that each partition receives an approximately equal number of messages, which is useful for load balancing and preventing hotspots.
| Property | Description | Values |
|---|---|---|
| partitioner.class | Class name of the partitioner implementation | org.apache.kafka.clients.producer.RoundRobinPartitioner |
Example: Producer Configuration with Round-Robin Partitioner
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class);
Note: The
RoundRobinPartitioner is the default partitioner in Kafka if no custom partitioner is specified. It ensures even distribution of messages across partitions.
7. Creating Sticky Partitioner
A sticky partitioner in Kafka ensures that messages with the same key are consistently routed to the same partition, while still allowing for load balancing across partitions. This is particularly useful in scenarios where you want to maintain message ordering within a key while distributing the load.
| Property | Description | Default Value |
|---|---|---|
| partitioner.class | Class name of the partitioner to use. For sticky partitioning, this is typically set to org.apache.kafka.clients.producer.StickyPartitioner. |
org.apache.kafka.clients.producer.RoundRobinPartitioner |
| partitioner.availability.timeout.ms | Timeout for partition availability check in milliseconds. If a partition is not available within this time, the producer may fall back to another strategy. | 0 |
Example: Using Sticky Partitioner in Producer Configuration
Properties props = new Properties();
props.put("bootstrap.servers", "localhost: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", "org.apache.kafka.clients.producer.StickyPartitioner");
Note: The
StickyPartitioner is the default partitioner in Kafka 2.0 and later versions, ensuring that messages with the same key are consistently routed to the same partition.
Key Features of Sticky Partitioner
- Ensures consistent partitioning for messages with the same key.
- Improves performance by reducing partition switching overhead.
- Supports dynamic partition availability checks.
8. Implementing Range Partitioner
| Property | Value | Description |
|---|---|---|
| partitioner.class | org.apache.kafka.clients.producer.internals.DefaultPartitioner |
Default partitioner class used if not specified |
| partitioner.class | org.apache.kafka.clients.producer.internals.RangePartitioner |
Custom range-based partitioner implementation |
Example: Range Partitioner Implementation
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
public class RangePartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
int numPartitions = cluster.partitionCountForTopic(topic);
int partitionId = Math.abs(key.hashCode()) % numPartitions;
return partitionId;
}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) {}
}
Note: The
RangePartitioner uses a simple modulo operation on the hash of the key to determine partition distribution.
Key Features of Range Partitioner
- Deterministic partitioning based on key hash
- Even distribution across partitions when keys are uniformly distributed
- Simple implementation suitable for basic use cases
9. Using Composite Key Partitioning
Composite key partitioning in Kafka allows you to distribute messages across partitions based on multiple fields within the message key. This is useful when a single field isn't sufficient to ensure even distribution or when you want to group related messages together.
| Field | Description | Example |
|---|---|---|
| key.field1 | First part of the composite key | "user_id" |
| key.field2 | Second part of the composite key | "region" |
| partitioner.class | Custom partitioner class to implement composite key logic | com.example.CompositeKeyPartitioner |
Example: Custom Partitioner Implementation
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;
public class CompositeKeyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
String compositeKey = (String) key;
String[] parts = compositeKey.split(":");
String field1 = parts[0];
String field2 = parts[1];
int partitionCount = cluster.partitionCountForTopic(topic);
int hash1 = Math.abs(field1.hashCode());
int hash2 = Math.abs(field2.hashCode());
return (hash1 + hash2) % partitionCount;
}
@Override
public void configure(Map<String, ?> configs) {}
@Override
public void close() {}
}
Note: The composite key should be structured consistently across messages to ensure predictable partitioning behavior.
Warning: If the composite key is not evenly distributed, some partitions may become hotspots, leading to performance bottlenecks.
10. Handling Null Keys
| Property | Description | Default Value | Notes |
|---|---|---|---|
| key | Partition key used for partition calculation | null |
If null, the partitioner must handle it gracefully |
Example: Custom Partitioner with Null Key Handling
public class NullKeyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
if (key == null || keyBytes == null || keyBytes.length == 0) {
// Handle null keys by assigning to a random partition
return ThreadLocalRandom.current().nextInt(cluster.partitionsForTopic(topic).size());
}
// Use standard hash-based partitioning for non-null keys
return Utils.toPositive(Utils.murmur2(keyBytes)) % cluster.partitionsForTopic(topic).size();
}
@Override
public void close() {}
@Override
public void configure(Map<String, ?> configs) {}
}
Note: When
key is null, the partitioner should either assign to a random partition or use a fallback strategy to ensure message distribution.