Skip to content

Purpose and scope of this Document

This document provides an overview of consuming data from a Kafka topic using one of the Kafka clients in Python. It aims to guide readers through the configuration and execution of a Kafka consumer 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 the playground environment, you can always use pkc-zd997.eu-central-1.aws.confluent.cloud:9092 - api_key = 'YOUR_API_KEY' # from API key management, "API secret" which is shared by the KaaS team - api_secret = 'YOUR_API_SECRET' # from API key management, "API secret" which is shared by the 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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
#Kafka configuration
conf = {
    'bootstrap.servers': bootstrap_server,
    'security.protocol': 'SASL_SSL',
    'sasl.mechanisms': 'PLAIN',
    'sasl.username': api_key,
    'sasl.password': api_secret,
    'group.id': 'sa-<namespace>-<env>-<optional postfix>',  # Follow our recommended consumer naming as per wiki under "consumer groups"
    'auto.offset.reset': 'latest'  # 'latest' or 'earliest'
}

#Create Consumer instance
consumer = Consumer(conf)

#Subscribe to topic
consumer.subscribe([topic_name])
print(f'Starting consumer... listening to topic: {topic_name}')

try:
    while True:
        msg = consumer.poll(2.0)  # Timeout in seconds
        if msg is None:
            continue

        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print('Reached end of partition')
            else:
                print(f'Error: {msg.error()}')
        else:
            print(f'Received message: {msg.value().decode("utf-8")}')
except KeyboardInterrupt:
    print('\nClosing consumer...')
finally:
    consumer.close()

Steps to run the Kafka Consumer 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_consumer.py. - Open your command line, navigate to the script's directory, and run it using python kafka_consumer.py. - The script has been tried and tested; it will begin listening for messages on the specified Kafka topic. - To stop the script, press Ctrl + C, allowing it to close gracefully by flushing any remaining messages. - Verify that messages were consumed successfully by checking the output in the terminal or reviewing the Kafka topic in the Confluent Cloud GUI.