AvroConsumer.consume() crash

[smoynes-tc]

Description

When calling AvroConsumer.consume(), Python crashes with a Segmentation fault.

I guess that this method should be overridden similar to AvroConsume.poll() to return deserialized messages.

How to reproduce

import confluent_kafka
import confluent_kafka.avro

c = confluent_kafka.avro.AvroConsumer(
    {
        "bootstrap.servers": ["kafka-broker"],
        "schema.registry.url": "http://schema-registry:8081",
    }
)

print(confluent_kafka.version(), confluent_kafka.libversion())
print(c.consume())

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): 0.11.5
• Apache Kafka broker version: 1.1.1
• Client configuration: {...}
• Operating system: Linux
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[rnpridgeon]

Thanks for reporting this as this is definitely a bug. As you are now aware the Avro Consumer does not actually support the consume method. That said it should have raised an AttributeError exception as opposed to segfaulting. We will look into fixing this soon and I have taken note of your interest in adding the consume attribute to the Avro Consumer implementation.

[smoynes-tc]

There's an implementation of AvroConsumer.consume() in #362. It looks like that PR needs review and, at least, some tests.

[rnpridgeon]

This will actually segfault for the standard Consumer as well. The fact that Avro does not overwrite this method is just a coincidence.

The problem is that you haven't set your group.id. As a result your consumer has not had its consumer queue initialized. This intern segfaults when the we try to access the queue upon calling consume.

We should raise an exception rather than segfaulting however.

[rnpridgeon]

Fixed with #506