Kafka Consumer Partition Assignment Strategies and Source Code Explanation

Kafka topics store data in partitions, each with a replication factor and a leader replica that consumers read from; this article focuses on how a consumer selects which partitions to consume and analyzes the core partition‑assignment strategies in Kafka.

Kafka supports two built‑in partition‑assignment strategies: the range strategy and the round‑robin strategy, which can be configured via the partition.assignment.strategy property when constructing a KafkaConsumer.

Using an example of a topic with eight partitions (p0‑p7) and three consumers (c0‑c2), the article demonstrates how each strategy distributes the partitions among the consumers.

Range strategy : partitions are divided sequentially; the number of partitions per consumer is n = pCount / cCount (8/3 = 2) with a remainder m = pCount % cCount (8%3 = 2). The first m consumers receive n+1 partitions, the rest receive n. The resulting allocation is illustrated with a diagram and the default configuration is shown in ConsumerConfig (RangeAssignor).

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, List<String>> subscriptions) {
    //获取每个主题消费者们
    Map<String, List<String>> consumersPerTopic = consumersPerTopic(subscriptions);
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());
    for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
        String topic = topicEntry.getKey();
        List<String> consumersForTopic = topicEntry.getValue();
        Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
        if (numPartitionsForTopic == null)
            continue;
        Collections.sort(consumersForTopic);
        int numPartitionsPerConsumer = numPartitionsForTopic / consumersForTopic.size();
        int consumersWithExtraPartition = numPartitionsForTopic % consumersForTopic.size();
        List<TopicPartition> partitions = partitions(topic, numPartitionsForTopic);
        for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
            int start = numPartitionsPerConsumer * i + Math.min(i, consumersWithExtraPartition);
            int length = numPartitionsPerConsumer + (i + 1 > consumersWithExtraPartition ? 0 : 1);
            assignment.get(consumersForTopic.get(i)).addAll(partitions.subList(start, start + length));
        }
    }
    return assignment;
}

Round‑robin strategy : partitions are assigned in a circular fashion across consumers (c0, c1, c2, c0, …). The article shows the step‑by‑step allocation (p0→c0, p1→c1, …) and provides a diagram of the resulting distribution.

@Override
public Map<String, List<TopicPartition>> assign(Map<String, Integer> partitionsPerTopic,
                                                Map<String, List<String>> subscriptions) {
    Map<String, List<TopicPartition>> assignment = new HashMap<>();
    for (String memberId : subscriptions.keySet())
        assignment.put(memberId, new ArrayList<TopicPartition>());
    CircularIterator<String> assigner = new CircularIterator<>(Utils.sorted(subscriptions.keySet()));
    for (TopicPartition partition : allPartitionsSorted(partitionsPerTopic, subscriptions)) {
        final String topic = partition.topic();
        while (!subscriptions.get(assigner.peek()).contains(topic))
            assigner.next();
        assignment.get(assigner.next()).add(partition);
    }
    return assignment;
}

The article concludes by noting that developers can create custom partition‑assignment strategies by extending the AbstractPartitionAssignor class and implementing their own assign method, using the provided context of topics, partitions, and consumer groups.

Finally, readers are encouraged to like, share, and bookmark the article.