Method: Kafka::Partitioner#call

Defined in:
lib/kafka/partitioner.rb

#call(partition_count, message) ⇒ Integer

Assigns a partition number based on a partition key. If no explicit partition key is provided, the message key will be used instead.

If the key is nil, then a random partition is selected. Otherwise, a digest of the key is used to deterministically find a partition. As long as the number of partitions doesn't change, the same key will always be assigned to the same partition.

Parameters:

  • partition_count (Integer) —

    the number of partitions in the topic.

  • message (Kafka::PendingMessage) —

    the message that should be assigned a partition.

Returns:

  • (Integer) —

    the partition number.

Raises:

  • (ArgumentError) 


27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/kafka/partitioner.rb', line 27

def call(partition_count, message)
  raise ArgumentError if partition_count == 0

  # If no explicit partition key is specified we use the message key instead.
  key = message.partition_key || message.key

  if key.nil?
    rand(partition_count)
  else
    @digest.hash(key) % partition_count
  end
end