Publish Messages to Kafka Topic from Python

Publish messages to Kafka Topic from Python using Confluent Kafka

In this post, I will show you how to publish messages to a Kafka Topic from Python using Confluent Kafka.

Prerequisites

• Kafka Cluster
• Kafka Topic
• Python

Install Confluent Kafka package

Install the confluent-kafka package using the following command:

pip install confluent-kafka

Publish messages to Kafka Topic

Add the following code in publish_messages_to_kafka_topic.py file to publish messages to the Kafka Topic:

from confluent_kafka import Producer

def delivery_report(err, msg):
    if err is not None:
        print(f'Message delivery failed: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

def publish_messages_to_topic(bootstrap_servers, topic_name):
    producer = Producer({'bootstrap.servers': bootstrap_servers})

    for i in range(10):
        producer.produce(topic_name, f'Message {i}', callback=delivery_report)

    producer.flush()

if __name__ == '__main__':
    bootstrap_servers = 'localhost:9092'
    topic_name = 'test-topic'

    publish_messages_to_topic(bootstrap_servers, topic_name)

Run the Python script

Run the Python script using the following command:

python publish_messages_to_kafka_topic.py

Using tools like offset explorer, you can verify that the messages are published to the Kafka Topic.

That’s it! You have successfully published messages to a Kafka Topic from Python using Confluent Kafka.

Things to remember

• Always handle message delivery reports to ensure that messages are successfully published to the Kafka Topic.
• Use the flush method to ensure that all messages are published before closing the producer.
• Use the callback parameter to handle message delivery reports.
• Use the bootstrap.servers parameter to specify the Kafka broker addresses.
• Do not create multiple producer instances if you are running your application in a loop. Instead, create a single producer instance and reuse it for publishing messages.