Skip to main content

Confluent Python client

The confluent-kafka Python package is a binding on top of the C client librdkafka. It comes bundled with a pre-built version of librdkafka.


The library needs Python 3.8+ installed. You can find installer here. You can install the Kafka-Python library either via PyPi, using pip, or by building the library from source. Installation documentation of pip.


You can instantiate a Producer object using a bunch of parameters.

The following example creates a Python producer for the <c8globals or c8locals>.<my-topic> topic and sends message on that topic.

from confluent_kafka import Producer

producer = Producer({'bootstrap.servers': '<my-gdn>.<my-paas>',
'': "confluent-client-1.7",
'sasl.username': '<my-tenant>/<my-fabric>'
'sasl.password': 'token:<my-api-key or my-JWT>',
'security.protocol': 'SASL_SSL',
'enable.ssl.certificate.verification': 'false',
'': '<path-to-ca.cet.pem-file>',
'sasl.mechanism': 'PLAIN'})

def acked(err, msg):
if err is not None:
print("Failed to deliver message: %s: %s" % (str(msg), str(err)))
print("Message produced: %s" % (str(msg)))

producer.produce('<c8globals or c8locals>.<my-topic>', key="<my-key-1>", value="<my-value-1>", callback=acked)

# Wait up to 1 second for events. Callbacks will be invoked during
# this method call if the message is acknowledged.


The following example creates a consumer with the <my-subscription> subscription name on the <c8globals or c8locals>.<my-topic> topic or multiple topics with global either local distribution. For example: consumer.subscribe(["c8globals.topic-1", "c8locals.topic-2"]). The listener receives incoming messages, prints the content and ID of messages that arrive.

from confluent_kafka import Consumer, KafkaException, KafkaError

consumer = Consumer({'bootstrap.servers': '<my-gdn>.<my-paas>',
'': config[""],
'auto.offset.reset': 'earliest',
'sasl.username': '<my-tenant>/<my-fabric>',
'sasl.password': 'token:<my-api-key or my-JWT>',
'security.protocol': 'SASL_SSL',
'enable.ssl.certificate.verification': 'false',
'': '<path-to-ca.cet.pem-file>',
'sasl.mechanism': 'PLAIN'})
consumer.subscribe([<c8globals or c8locals>.<my-topic>])

while True:
msg = consumer.poll(timeout=1.0)
if msg is None: continue

if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
# End of partition event
print('topic {} in partition {} reached end at offset {}'.format(msg.topic(), msg.partition(), msg.offset()))
elif msg.error():
raise KafkaException(msg.error())
print("message.offset={}, message.key={}, message.value={}".format(msg.offset(), msg.key(), msg.value()))