key为null时Kafka会将消息发送给哪个分区?
当你编写kafka Producer时， 会生成KeyedMessage对象。
| 1 | KeyedMessage<K, V> keyedMessage = new KeyedMessage<>(topicName, key, message) |
这里的key值可以为空，在这种情况下， kafka会将这个消息发送到哪个分区上呢？依据Kafka官方的文档， 默认的分区类会随机挑选一个分区：
The third property "partitioner.class" defines what class to use to determine which Partition in the Topic the message is to be sent to. This is optional, but for any non-trivial implementation you are going to want to implement a partitioning scheme. More about the implementation of this class later. If you include a value for the key but haven't defined a partitioner.class Kafka will use the default partitioner. If the key is null, then the Producer will assign the message to a random Partition.
但是这句话相当的误导人。
从字面上来讲，这句话没有问题， 但是这里的随机是指在参数"topic.metadata.refresh.ms"刷新后随机选择一个， 这个时间段内总是使用唯一的分区。 默认情况下每十分钟才可能重新选择一个新的分区。 但是相信大部分的程序员和我一样， 都理解成每个消息都会随机选择一个分区。可以查看相关的代码：key为null时Kafka会将消息发送给哪个分区?
当你编写kafka Producer时， 会生成KeyedMessage对象。
| 1 | KeyedMessage<K, V> keyedMessage = new KeyedMessage<>(topicName, key, message) |
这里的key值可以为空，在这种情况下， kafka会将这个消息发送到哪个分区上呢？依据Kafka官方的文档， 默认的分区类会随机挑选一个分区：
The third property "partitioner.class" defines what class to use to determine which Partition in the Topic the message is to be sent to. This is optional, but for any non-trivial implementation you are going to want to implement a partitioning scheme. More about the implementation of this class later. If you include a value for the key but haven't defined a partitioner.class Kafka will use the default partitioner. If the key is null, then the Producer will assign the message to a random Partition.
但是这句话相当的误导人。
从字面上来讲，这句话没有问题， 但是这里的随机是指在参数"topic.metadata.refresh.ms"刷新后随机选择一个， 这个时间段内总是使用唯一的分区。 默认情况下每十分钟才可能重新选择一个新的分区。 但是相信大部分的程序员和我一样， 都理解成每个消息都会随机选择一个分区。可以查看相关的代码：| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 | private def getPartition(topic: String, key: Any, topicPartitionList: Seq[PartitionAndLeader]): Int = { val numPartitions = topicPartitionList.size if(numPartitions <= 0) throw new UnknownTopicOrPartitionException("Topic " + topic + " doesn't exist") val partition = if(key == null) { // If the key is null, we don't really need a partitioner // So we look up in the send partition cache for the topic to decide the target partition val id = sendPartitionPerTopicCache.get(topic) id match { case Some(partitionId) => // directly return the partitionId without checking availability of the leader, // since we want to postpone the failure until the send operation anyways partitionId case None => val availablePartitions = topicPartitionList.filter(_.leaderBrokerIdOpt.isDefined) if (availablePartitions.isEmpty) throw new LeaderNotAvailableException("No leader for any partition in topic " + topic) val index = Utils.abs(Random.nextInt) % availablePartitions.size val partitionId = availablePartitions(index).partitionId sendPartitionPerTopicCache.put(topic, partitionId) partitionId } } else partitioner.partition(key, numPartitions) if(partition < 0 || partition >= numPartitions) throw new UnknownTopicOrPartitionException("Invalid partition id: " + partition + " for topic " + topic + "; Valid values are in the inclusive range of [0, " + (numPartitions-1) + "]") trace( "Assigning message of topic %s and key %s to a selected partition %d".format(topic, if (key == null) "[none]" else key.toString, partition)) partition } |
如果key为null, 它会从sendPartitionPerTopicCache查选缓存的分区， 如果没有，随机选择一个分区，否则就用缓存的分区。
LinkedIn工程师Guozhang Wang在中解释了这一问题，
最初kafka是按照大部分用户理解的那样每次都随机选择一个分区， 后来改成了定期选择一个分区， 这是为了减少服务器段socket的数量。不过这的确很误导用户，据称0.8.2版本后又改回了每次随机选取。但是我查看0.8.2的代码还没看到改动。所以，如果有可能，还是为KeyedMessage设置一个key值吧。
LinkedIn工程师Guozhang Wang在中解释了这一问题，如果key为null, 它会从sendPartitionPerTopicCache查选缓存的分区， 如果没有，随机选择一个分区，否则就用缓存的分区。
最初kafka是按照大部分用户理解的那样每次都随机选择一个分区， 后来改成了定期选择一个分区， 这是为了减少服务器段socket的数量。不过这的确很误导用户，据称0.8.2版本后又改回了每次随机选取。但是我查看0.8.2的代码还没看到改动。
所以，如果有可能，还是为KeyedMessage设置一个key值吧。
from：http://colobu.com/2015/01/22/which-kafka-partition-will-keyedMessages-be-sent-to-if-key-is-null/