Kafka Avro Consumer Python

consumer.py

from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError

c = AvroConsumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'groupid',
    'schema.registry.url': 'http://127.0.0.1:8081'
})

c.subscribe(['reddit_posts'])

while True:
    try:
        msg = c.poll(10)
    except SerializerError as e:
        print("Message deserialization failed for {}: {}".format(msg, e))
        break

    if msg is None:
        continue

    if msg.error():
        print("AvroConsumer error: {}".format(msg.error()))
        continue

    value = msg.value()
    print(value["author"])

c.close()

requirements.txt

avro-python3==1.8.2
certifi==2019.3.9
chardet==3.0.4
confluent-kafka==0.11.6
idna==2.8
requests==2.21.0
urllib3==1.24.1