Purpose and scope of this Document

This document provides an overview of publishing data to a Kafka topic using one of the Kafka clients in Python. It aims to guide readers through the configuration and execution of a Kafka producer script, ensuring a clear understanding of the necessary steps involved in the process.

Pre-Read: Get started with different Kafka clients

Pre-Req Info: - bootstrap_server = 'YOUR_BOOTSTRAP_SERVER' # from cluster settings
Note: For playground env you can always use pkc-zd997.eu-central-1.aws.confluent.cloud:9092
- api_key = 'YOUR_API_KEY' # from API key management, "API key" which is shared by KaaS Team
- api_secret = 'YOUR_API_SECRET' # from API key management, "API secret" which is shared by KaaS Team
- topic_name = 'YOUR_TOPIC_NAME' # your topic name configured as per naming standard
- groupId/clientId = 'GROUPID/CLIENTID' # It is mandatory to have a prefix as "sa-[yourserviceaccountname]-" for groupID/ClientID

Python Script

# Kafka configuration
conf = {
    'bootstrap.servers': bootstrap_server,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'PLAIN',
    'sasl.username': api_key,
    'sasl.password': api_secret
}
# Create Producer instance
producer = Producer(conf)

def delivery_report(err, msg):
    """ Called once for each message produced to indicate delivery result.
    Triggered by `produce` or `flush`. """
    if err is not None:
        print(f'Error delivering message: {err}')
    else:
        print(f'Message delivered to {msg.topic()} [{msg.partition()}]')

try:
    # Specify the topic and start sending messages
    while True:
        message = input("Enter a message to send: ")  # Input message from user
        # Produce message to topic
        producer.produce(topic_name, key=None, value=message, callback=delivery_report)
        # Wait up to 1 second for events. Callbacks will be invoked during
        # produce() call if delivery reports are enabled.
        producer.poll(1.0)
except KeyboardInterrupt:
    print('\nClosing producer...')
finally:
    # Flush any remaining messages in the Producer queue
    producer.flush()

Steps to run Kafka producer script - Install the confluent_kafka library using pip install confluent-kafka.
- Replace the placeholders in the script with your actual bootstrap server URL, API key, API secret, and topic name.
- Save the configured script with a .py extension, such as kafka_producer.py.
- Open your command line, navigate to the script's directory, and run it using python kafka_producer.py.
- Enter messages at the prompt to send them to the Kafka topic.
- To stop the script, press Ctrl + C, allowing it to flush any remaining messages before exiting.
- Verify the messages were published successfully by consuming them from the topic or checking the Confluent Cloud GUI.