retrieve messages in a topic using kafka-python

Question

I have written a python script using kafka-python library which writes and reads messages into kafka. I write messages without any problem; I can retrieve them using kafka console tools. But I can't read them using my python script. I have a for on my consumer which freezes on the first line of the iteration and never returns. Here's my code:

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "my-topic",
    bootstrap_servers="localhost:9092"),
    value_deserializer=lambda v: json.dumps(v).encode("utf-8")
)

for msg in consumer:
    print(type(msg))

The consumer is created and subscribed completely; I can see that my-topic is listed on the topic list of its _client property.

Any idea?

Answer 1

By default, kafka python start from last offset, ie will only read new mesages. One approach is to read from beginning or the alternative approach is to keep polling topic in an infinite loop as shown in below code:

while True:
    try:
        records = consumer.poll(60 * 1000) # timeout in millis , here set to 1 min

        record_list = []
        for tp, consumer_records in records.items():
            for consumer_record in consumer_records:
                record_list.append(consumer_record.value)
        print(record_list) # record_list will be list of dictionaries

Edit

To read from the beginning, we need to add auto_offset_reset=earliest earlies while making consumer object

consumer = KafkaConsumer(
    "my-topic",
    bootstrap_servers="localhost:9092"),
    value_deserializer=lambda v: json.dumps(v).encode("utf-8"),
    auto_offset_reset='earliest')

Let me know if this helps!!