ApsaraMQ for Kafka:Best practices for producers

Keyed messages

For messages with a key, the producer hashes the key and selects a partition based on the hash result. Messages with the same key always go to the same partition.

Keyless messages

Before Kafka 2.4, the default strategy for keyless messages was round-robin: each message went to the next partition in sequence. This scattered messages across all partitions, producing many small batches and increasing latency.

Kafka 2.4 introduced the sticky partitioning strategy (KIP-480) to solve this problem. Instead of rotating partitions per message, the producer sticks to one partition until the current batch is full, then randomly selects a new partition. Over time, messages are still evenly distributed across all partitions, but batches are much larger. This approach prevents message partition skew, reduces latency, and improves overall service performance.

Enable sticky partitioning

• Client version 2.4 or later: Sticky partitioning is the default. No configuration needed.

• Client version earlier than 2.4: Implement a custom partitioner and set it via partitioner.class. The following example switches partitions at a configurable time interval:

public class MyStickyPartitioner implements Partitioner {

    // Records the time of the last partition switch.
    private long lastPartitionChangeTimeMillis = 0L;
    // Records the current partition.
    private int currentPartition = -1;
    // The partition switch interval. Set the interval as needed.
    private long partitionChangeTimeGap = 100L;

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

    /**
     * Compute the partition for the given record.
     *
     * @param topic The topic name
     * @param key The key to partition on (or null if no key)
     * @param keyBytes serialized key to partition on (or null if no key)
     * @param value The value to partition on or null
     * @param valueBytes serialized value to partition on or null
     * @param cluster The current cluster metadata
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        // Get all partition information.
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null) {
            List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
            int availablePartitionSize = availablePartitions.size();

            // Check the current active partitions.
            if (availablePartitionSize > 0) {
                handlePartitionChange(availablePartitionSize);
                return availablePartitions.get(currentPartition).partition();
            } else {
                handlePartitionChange(numPartitions);
                return currentPartition;
            }
        } else {
            // For a message with a key, select a partition based on the key's hash value.
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }

    private void handlePartitionChange(int partitionNum) {
        long currentTimeMillis = System.currentTimeMillis();

        // If the time since the last switch exceeds the switch interval, switch to the next partition. Otherwise, use the current partition.
        if (currentTimeMillis - lastPartitionChangeTimeMillis >= partitionChangeTimeGap
            || currentPartition < 0 || currentPartition >= partitionNum) {
            lastPartitionChangeTimeMillis = currentTimeMillis;
            currentPartition = Utils.toPositive(ThreadLocalRandom.current().nextInt()) % partitionNum;
        }
    }

    public void close() {}

}