Python deserialize kafka message with avro repository

Question

I need to read Kafka messages using an avro stored in the repository. Using kafka-python 2.0.2, I can connect to the Kafka topic and read the messages but I have no idea on how to decode them.

from kafka import KafkaConsumer
consumer = KafkaConsumer('SOME-TOPIC', 
                        other connection parameters,
                        auto_offset_reset= 'earliest')
                        # value_deserializer=lambda m: json.loads(m.decode('utf-8')))
                        # value_deserializer=lambda m: decode(m))

for msg in consumer:
    print (msg)

What library should I use ? confluent-kafka 1.5.0, avro-python3 1.10.1 How to proceed ?

Identify the version of the message
Connect to the avro repository
Get the avro with the right version
Use it to decode the message

That seems a lot to do, is there not a simpler way to do ? I would appreciate to get an example to guide me.

To connect to the avro repository I have these parameters

basic.auth.credentials.source
schema.registry.basic.auth.user.info
schema.registry.url

GuyOlivier GuyOlivier · Accepted Answer · 2021-01-26T13:48:44

I got it working with the following libraries:

pip install confluent-avro
pip install kafka-python

And the code:

from kafka import KafkaConsumer

from confluent_avro import AvroKeyValueSerde, SchemaRegistry
from confluent_avro.schema_registry import HTTPBasicAuth

KAFKA_TOPIC = "SOME-TOPIC"

registry_client = SchemaRegistry(
    "https://...",
    HTTPBasicAuth("USER", "PASSWORD"),
    headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
)
avroSerde = AvroKeyValueSerde(registry_client, KAFKA_TOPIC)

consumer = KafkaConsumer(KAFKA_TOPIC, 
                        other connection parameters,
                        auto_offset_reset= 'earliest')

for msg in consumer:
    v = avroSerde.value.deserialize(msg.value)
    k = avroSerde.key.deserialize(msg.key)
    print(msg.offset, msg.partition, k, v)
    break

Reference: https://pypi.org/project/confluent_avro/

Python deserialize kafka message with avro repository

1 Answers