Source code for kafka.producer.keyed

from __future__ import absolute_import

import logging
import warnings

from kafka.producer.base import Producer
from kafka.partitioner import HashedPartitioner


log = logging.getLogger(__name__)


[docs]class KeyedProducer(Producer): """ A producer which distributes messages to partitions based on the key See Producer class for Arguments Additional Arguments: partitioner: A partitioner class that will be used to get the partition to send the message to. Must be derived from Partitioner. Defaults to HashedPartitioner. """ def __init__(self, *args, **kwargs): self.partitioner_class = kwargs.pop('partitioner', HashedPartitioner) self.partitioners = {} super(KeyedProducer, self).__init__(*args, **kwargs) def _next_partition(self, topic, key): if topic not in self.partitioners: if not self.client.has_metadata_for_topic(topic): self.client.load_metadata_for_topics(topic, ignore_leadernotavailable=True) self.partitioners[topic] = self.partitioner_class(self.client.get_partition_ids_for_topic(topic)) partitioner = self.partitioners[topic] return partitioner.partition(key)
[docs] def send_messages(self, topic, key, *msg): partition = self._next_partition(topic, key) return self._send_messages(topic, partition, *msg, key=key)
# DEPRECATED
[docs] def send(self, topic, key, msg): warnings.warn("KeyedProducer.send is deprecated in favor of send_messages", DeprecationWarning) return self.send_messages(topic, key, msg)
def __repr__(self): return '<KeyedProducer batch=%s>' % (self.async_send,)