KafkaProducer.doSend：Partitioner的双重调用导致分配不均

Question

版本：

• 弹簧靴：2.4.1
• Kafka 客户端：2.6.0

由于某些系统限制，Kafka Producer 被配置为单独发送每条消息。具体来说，通过设置linger.ms = 0 和max.batch.size = 0 禁用批处理。 因此，从 Kafka 的客户端角度来看，每条消息都是新批次。我想在所有可用分区之间平均分配消息，这就是我配置客户端RoundRobinPartitioner 来实现它的原因。

但是，当一个节点关闭并且我有偶数个分区时对配置进行测试，我发现消息分布在所有可用分区的一半之间。这种行为的原因是在KafkaProducer.doSend(..)'中双重调用partitioner.partition(..)。由于 RoundRobinPartitioner.partition() 在后台递增计数器并返回除以可用分区号的余数，因此在一个记录发布中调用它两次会导致跳过每个第二个分区。

例如，availablePartitions 包含 6 个分区 (1-6)。通过KafkaProducer.doSend(..) 两次调用partition(..) 总是跳过1,3,5 个分区。

 private Future<RecordMetadata> doSend(ProducerRecord<K, V> record, Callback callback) {
 ..
 int partition = partition(record, serializedKey, serializedValue, cluster);
 ..
 RecordAccumulator.RecordAppendResult result = accumulator.append(tp, timestamp, serializedKey,
                serializedValue, headers, interceptCallback, remainingWaitMs, true, nowMs);

//Since I disabled batches this if is always true
        if (result.abortForNewBatch) {
            int prevPartition = partition;
            partitioner.onNewBatch(record.topic(), cluster, prevPartition);
            partition = partition(record, serializedKey, serializedValue, cluster);
            tp = new TopicPartition(record.topic(), partition);
            if (log.isTraceEnabled()) {
                log.trace("Retrying append due to new batch creation for topic {} partition {}. The old partition was {}", record.topic(), partition, prevPartition);
            }
            // producer callback will make sure to call both 'callback' and interceptor callback
            interceptCallback = new InterceptorCallback<>(callback, this.interceptors, tp);

            result = accumulator.append(tp, timestamp, serializedKey,
                serializedValue, headers, interceptCallback, remainingWaitMs, false, nowMs);
        }

1. 虽然很容易重新实现，但我想知道这种行为是否 想要？
2. 如果当前批次更新分区的原因是什么 关门了吗？
3. 我相信我不是第一个，How 社区 克服了这种设计？

Answer 1

可以使用UniformStickyPartitioner 作为解决方法。它完全符合需要：仅在newBatch(..) 调用时选择新分区，而对partition(..) 的多次调用会产生相同的输出。真正的 Round Robin 机制的唯一区别是 UniformStickyPartitioner 策略是分区计算为 randomInt % availablePartitions。但是，分配仍然相当平均。

Answer 2

是的，我之前也遇到过这个问题，让我困惑了很久。

Google 上对此几乎没有答案，Kafka 社区也没有解决这个问题。

但是Apache Jira上有相关讨论：

KAFKA-13303

KAFKA-9965

KAFKA-9964