Hey Flink Community, I'm working on developing a PyFlink Consumer that reads Avro Formatted messages from Kafka, and I'm having some issues deserializing messages. So far I've tried constructing a FlinkKafkaConsumer using an AvroRowDeserializationSchema constructed from my Avro Schema like so:

with open("/opt/flink/usrlib/LogRecord.avsc","r") as f:
    avro_schema_string=f.read()
deserialization_schema = AvroRowDeserializationSchema(
    avro_schema_string=avro_schema_string
)
kafka_consumer = FlinkKafkaConsumer(
    topic="MyTopic",
    deserialization_schema=deserialization_schema,
    properties={
        'bootstrap.servers': f'{host}:{port},
        'group.id': my_consumer_group,

Flink is able to connect, but I'm getting errors like so:

Caused by: org.apache.avro.AvroRuntimeException: Malformed data. Length is negative: -1
	at org.apache.avro.io.BinaryDecoder.readString(BinaryDecoder.java:308)
	at org.apache.avro.io.ResolvingDecoder.readString(ResolvingDecoder.java:208)
	at org.apache.avro.generic.GenericDatumReader.readString(GenericDatumReader.java:470)

I know the messages are valid, because they were serialized using the same schema and avro version before going into Kafka. Does anyone have any suggestions? I'm new to the JVM, so I suspect that the fat jar I made to hold all my project dependencies isn't following best practices.

Do you use a Schema Registry when producing messages?

+1 on the first re-question - your first few bytes might be apart of the envelope

Do you use a Schema Registry when producing messages?

@sap1ens, We do, but I'm not sure how to configure flink to consume from it.

👍 no worries, just wanted to make sure first. As Marquis pointed, when using the Schema Registry, the first 4 bytes are used to encode the schema id, that’s why you see the exception. Of course, you can modify the payload, skip the first 4 bytes, but then you won’t get any benefits of using the Schema Registry. Consider using

RegistryAvroDeserializationSchema

instead of

AvroRowDeserializationSchema

. This requires providing a Schema Registry client via

SchemaCoderProvider

, but I don’t have an example in Python for that.

Thanks Both of y'all for the info, that makes prefect sense about the wire format. Does PyFlink support

RegistryAvroDeserializationSchema

at all? I'm looking around in pyflink.datastream.formats.avro and I don't see any modules or classes that look like they reach out the JVM. I suppose I could implement my own to test.

I use Table API on PyFlink to be able to use the Confluent Schema Registry with the Wire Format. I haven't found a way to use DataStream API with the Schema Registry on PyFlink.