一、Producer Partitioner 不均是什么原因？
由于社区推荐 UniformStickyPartitioner 分区器作为默认分区器，但是由于 Kafka 开源 Java SDK，版本>=2.4.1 && < 3.3.0，其中 Kafka Producer 默认的粘性分区策略存在数据倾斜问题，尤其 linger.ms 为0（短时间的连续消息均被发送至同一个分区内），客户如果对热点数据倾斜较为敏感的场景，可以考虑 Kafka Java SDK 升级到3.3.0以及以上，Kafka Java SDK Producer 在3.3.0开发了 Strictly Uniform Sticky Partitioner 对此进行优化。高版本：写入一定要指明 ack，幂等，linger.ms。
/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */package org.apache.kafka.clients.producer;import java.util.List;import java.util.Map;import java.util.Queue;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.ConcurrentLinkedQueue;import java.util.concurrent.ConcurrentMap;import java.util.concurrent.atomic.AtomicInteger;import org.apache.kafka.common.Cluster;import org.apache.kafka.common.PartitionInfo;import org.apache.kafka.common.utils.Utils;import org.slf4j.Logger;import org.slf4j.LoggerFactory;/** * The "Round-Robin" partitioner – MODIFIED TO WORK PROPERLY WITH STICKY PARTITIONING (KIP-480) * <p> * This partitioning strategy can be used when user wants to distribute the writes to all * partitions equally. This is the behaviour regardless of record key hash. */public class MyRoundRobinPartitioner implements Partitioner { private static final Logger LOGGER = LoggerFactory.getLogger(RoundRobinPartitioner.class); private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap<>(); private final ConcurrentMap<String, Queue<Integer>> topicPartitionQueueMap = new ConcurrentHashMap<>(); 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 */ @Override public int partition( String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { Queue<Integer> partitionQueue = partitionQueueComputeIfAbsent(topic); Integer queuedPartition = partitionQueue.poll(); if (queuedPartition != null) { LOGGER.trace("Partition chosen from queue: {}", queuedPartition); return queuedPartition; } else { List<PartitionInfo> partitions = cluster.partitionsForTopic(topic); int numPartitions = partitions.size(); int nextValue = nextValue(topic); List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic); if (!availablePartitions.isEmpty()) { int part = Utils.toPositive(nextValue) % availablePartitions.size(); int partition = availablePartitions.get(part).partition(); LOGGER.trace("Partition chosen: {}", partition); return partition; } else { // no partitions are available, give a non-available partition return Utils.toPositive(nextValue) % numPartitions; } } } private int nextValue(String topic) { AtomicInteger counter = topicCounterMap.computeIfAbsent( topic, k -> { return new AtomicInteger(0); }); return counter.getAndIncrement(); } private Queue<Integer> partitionQueueComputeIfAbsent(String topic) { return topicPartitionQueueMap.computeIfAbsent(topic, k -> { return new ConcurrentLinkedQueue<>(); }); } public void close() { } /** * Notifies the partitioner a new batch is about to be created. When using the sticky partitioner, * this method can change the chosen sticky partition for the new batch. * * @param topic The topic name * @param cluster The current cluster metadata * @param prevPartition The partition previously selected for the record that triggered a new * batch */ @Override public void onNewBatch(String topic, Cluster cluster, int prevPartition) { LOGGER.trace("New batch so enqueuing partition {} for topic {}", prevPartition, topic); Queue<Integer> partitionQueue = partitionQueueComputeIfAbsent(topic); partitionQueue.add(prevPartition); }}
二、为什么 Consumer coordinator 无法刷新重连？
Kafka Java consumer 客户端，版本>=2.6.x&&<3.2.1，在 commitOffsetsAsync 异常后，无法恢复与 coordinator 的连接，从而导致后续无法提交位点，该 issue 在3.2.1以及之后的版本修复，建议存在该问题的客户端，可要么升级到3.2.1进行解决，要么回滚到<2.6.x版本。
三、为什么 Consumer 粘性分配器重复消费？
Kafka Java consumer 客户端<2.3.0版本，如果使用 Sticky assignor，会出现分区重复分配，即一个分区可能同时分配给同一个消费组的不同消费者消费，从而造成重复消费问题，如果使用 Sticky assignor，建议客户调整版本到2.3.0以及以上，避免该问题。