Kafka Producers¶
The Confluent Platform includes the standard Java producer shipped with Kafka itself, the C/C++ client library librdkafka, the the Python client library confluent-kafka-python. This section gives a high-level overview of how the producer works, an introduction to the configuration settings for tuning, and some examples from each client library.. These settings are
the same for Java, C/C++, and Python++
and Python clients, this is a per-topic configuration, but can be applied
globally using the
default_topic_conf sub-configuration in C/C++ and
default.topic.config sub-configuration in Python. C/C++ and Python clients
the setting is
message.send.max.retries and the default is 2.++ and Python clients, you can use
batch.num.messages to set a limit on
the number of messages contained in each batch. To enable compression,
use
compression.codec.++ and Python clients have similar settings. Use
queue.buffering.max.messages
to limit the total number of unsent messages that can be queued at any
given time. To limit the amount of time that messages can remain in
the queue, use
queue.buffering.max.ms.); } rd_kafka_topic_conf_t *topic_conf = rd_kafka_topic_conf_new(); if (rd_kafka_topic_conf_set(topic_conf, "acks", "all", producer: %s\n", errstr); exit(1); }
from confluent_kafka import Producer import socket conf = {'bootstrap.servers': "host1:9092,host2:9092", 'client.id': socket.gethostname(), 'default.topic.config': {'acks': 'all'}} producer = Producer(conf), topic_conf);. Here we passed the
topic_conf we seeded
with a configuration for acknowledgments. Passing
NULL will cause
the producer to use.
In Python, you initiate a send by calling the
produce method, passing
in the value, and optionally a key, partition, and callback. The call will
return immediately and does not return a value.
producer.produce(topic, key="key", value="value")..
Finally,) C/C++ and Python using the
delivery callback, but it takes a bit more work. A full example can be
found here.
The Python client also contains a
flush() method which has the same
effect:
producer.produce(topic, key="key", value="value") producer.flush()