Python - Exit Kafka queue once all messages have been read

user_9776699 0
Viewed 516

I am trying to read some data of a Kafka queue using Python as shown in the below code:

from kafka import KafkaConsumer
import sys
import json 
import pandas as pd


bootstrap_servers = [localhost]
topicName = 'topic'
consumer = KafkaConsumer (topicName, group_id = 'topic',bootstrap_servers = bootstrap_servers, auto_offset_reset = 'earliest')

data_list = []
for message in consumer:
    print(message)
    data = json.loads(message.value)
    df = pd.json_normalize(data)
    data_list.append(df)

This seems to be running in a loop forever unless I kill the connection. Is there a way how I can stop/come out of this loop after having read all of the messages or there are no new messages in the queue?

kafka-consumer-api apache-kafka python
2 Answers

The poll method should be what you're looking for.

Just be aware of the max_records parameters defaulting to max_poll_records which is 500 records if not changed

user_76057620

You might need to close the consumer once the message consumption is done:

for message in consumer:
    print(message)
    data = json.loads(message.value)
    df = pd.json_normalize(data)
    data_list.append(df)

consumer.close()
user_71317570
Related