DeserializingConsumer does not return a regular Message when used with AvroDeserializer

[InterferencePattern]

Description

The Message object returned by DeserializingConsumer.value() doesn't appear to match the signature- it can return a dict, for example, and not only a str or bytes.

How to reproduce

1. Use AvroDeserializer to create DeserializingConsumer
2. Use .poll() method to return message
3. Message.value() should contain deserialized dictionary instead of str or bytes

I wonder if the DeserializingConsumer needs a DeserializedMessage, with something like the following:

class DeserializedMessage(Message):
    def value(self, payload=None) -> dict:
        return super().value(payload)

Checklist

Please provide the following information:

• [x] confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): ('2.3.0', 33751040), ('2.3.0', 33751295)
• [ ] Apache Kafka broker version:
• [ ] Client configuration: {...}
• [ ] Operating system:
• [ ] Provide client logs (with 'debug': '..' as necessary)
• [ ] Provide broker log excerpts
• [ ] Critical issue

[InterferencePattern]

@pranavrth might be the contributor most familar with the DeserializingConsumer

[pranavrth]

You need to use key.deserializer or value.deserializer config to make it work with deserializer otherwise you will get the message in str or bytes.

[InterferencePattern]

That's exactly what I'm saying. This issue isn't that it doesn't work- it does.

But the Message.value() return from a DeserializingConsumer is a dictionary, right? If so, this doesn't conform to the output in the Message documentation and should be its own class.

I would offer to define such a class (like in the original description above), but the original Message class is not defined in Python.

[pranavrth]

But the Message.value() return from a DeserializingConsumer is a dictionary, right?

No, it depends on the value serializer that is provided. If no serializer is provided, it is same as what a normal consumer would return i.e. byte/str.

[InterferencePattern]

Okay, I've narrowed the Issue title to indicate that this is true for the AvroDeserializer. It's really just a matter of changing the expected output type to be more inclusive of the possible values.

[pranavrth]

I would let @rayokota handle from here whose team deals with Schema Registry.