Avro consumer return schema #453

saabeilin · 2018-09-20T09:30:09Z

That's a second attempt to get the schema itself together with the decoded Avro message.
The motivation is described in #449

The previous approach was expecting the decoded message to be a dictionary which is not always true, for instance, it's quite common to have avro-encoded strings as keys. Here we create the wrapping class dynamically, subclassing from the type of the decoded value and adding the schema mixin.

…o schema

…on into avro-consumer-return-schema

`avro` and `avro-python3` are not 100% compatible. For instance, `PrimitiveSchema` class in python3 has the `name` property while the python2-version does not.

ghost · 2018-09-20T09:30:14Z

@confluentinc It looks like @saabeilin just signed our Contributor License Agreement. 👍

Always at your service,

clabot

rnpridgeon · 2018-09-20T11:15:17Z

Thanks again @saabeilin!

There are two concerns with this current approach:

It can't be turned off
It explodes the application's memory footprint significantly
- For each Message we potentially need to allocate storage for the Message key and value schema. This schema is already cached by the CachedSchemaRegistryClient and shared across messages.

That said I think a better approach would be to enable the application to query the CachedSchemaRegistryClient's cache via some function. This has three major advantages:

It's optional, user's only have to pay for the look up if they call the function
It reduces schema storage duplication
It brings the CachedSchemaRegistryClient closer to feature parity with the Java client 👍

saabeilin · 2018-09-20T11:40:17Z

Yes, and it's not optional in Java, for instance.
Actually not that much. Before the object is modified, only the reference will be kept. It's easy to check by comparing schemas for two message with is. I did a brief check (we use an additional iterator-like wrapper around the consumer)

In [10]: from notifications.consumer import consumer

In [11]: m1 = next(consumer)

In [12]: m2 = next(consumer)

In [13]: m1 == m2, m1 is m2
Out[13]: (False, False)

In [14]: m1.value._schema == m2.value._schema, m1.value._schema is m2.value._schema
Out[14]: (True, True)

So from my point of view it's a rather cheap operation. In many situations the wrappers around will build some "real" Python objects based on schema and the dictionary.

The Registry client, itself, should be out of interest for most of applied developers. We deal with business logic and the registry client is something working under the hood, we just provide a schema so messages are serialized, and get the schema back.

rnpridgeon · 2018-09-21T13:26:45Z

Ah right python passes immutable objects by reference until they are mutated sorry about that. Right so with that in mind I think your right this should work.

Can you add unit tests which cover wrapping primitive and record types. We should also add documentation for this feature.

saabeilin · 2018-09-24T20:24:36Z

@rnpridgeon Ryan, do you think these tests are enough? What/where shall we put these in the docs?

edenhill · 2018-09-25T19:29:54Z

confluent_kafka/avro/serializer/message_serializer.py

@@ -59,6 +59,30 @@ def __exit__(self, *args):
        return False


+class HasSchemaMixin:
+    """
+    Wraps a decoded Avro message dictionary to make able to add schema


change to ".. to add schema attribute"

confluent_kafka/avro/serializer/message_serializer.py

edenhill · 2018-09-25T19:31:17Z

confluent_kafka/avro/serializer/message_serializer.py

+    else:
+        name = schema.type
+
+    klass = type(str(name), (value.__class__, HasSchemaMixin), {})


isn't name already a string?

klass is swedish, use cls or similar.

If we still support 2.7, is not name a formally unicode object?

edenhill · 2018-09-25T19:31:29Z

confluent_kafka/avro/serializer/message_serializer.py

+
+
+def _wrap(value, schema):
+    if hasattr(schema, 'namespace'):


edenhill · 2018-09-25T19:35:06Z

confluent_kafka/avro/serializer/message_serializer.py

@@ -213,4 +237,5 @@ def decode_message(self, message):
            if magic != MAGIC_BYTE:
                raise SerializerError("message does not start with magic byte")
            decoder_func = self._get_decoder_func(schema_id, payload)
-            return decoder_func(payload)
+            decoded_message = _wrap(decoder_func(payload), self.registry_client.get_by_id(schema_id))


What are the performance implications of this change?
It would be good with comparative numbers between master and this fix.

Also, there is no need for the temporary decoded_message variable, should return _wrap ..

I did not do any performance testing, but my guess will be avro decoding will be the slowest part. On the other hand, in really many situations the message itself without the schema (or, at least, the full name of the schema which is usually the "message class") becomes close to useless.

edenhill · 2018-09-25T19:35:41Z

tests/avro/test_message_serializer.py

+
+    def test_schema_mixin_wrapper(self):
+        schema = avro.loads(data_gen.BASIC_SCHEMA)
+        for kls in (int, float, dict, list):


swedish klass

edenhill · 2018-09-25T19:40:30Z

confluent_kafka/avro/serializer/message_serializer.py

+    klass = type(str(name), (value.__class__, HasSchemaMixin), {})
+
+    wrapped = klass(value)
+    wrapped._schema = schema


In the unlikely, but possible, event that existing user's class has a ._schema field, this will silently overwrite it, and is as such a breaking change.

I agree, but I do not have a better solution. An option (though a breaking change) would be returning a tuple (value, schema) instead, does it make sense?

saabeilin · 2018-10-15T13:28:17Z

@rnpridgeon @edenhill Since you've closed #449 - I shall backport these changes to Kafkian. It currently relies on this branch for it to work.

rnpridgeon · 2018-10-15T13:34:44Z

Let's keep this open until @edenhill chimes in but I think in the end both he and I agreed this should be an extension of the C message type.

I actually thought you had already merged this into Kafkian so closing it was a bit premature. I'll reopen it until we come into a consensus.

See confluentinc/confluent-kafka-python#453

soxofaan · 2018-10-26T09:52:14Z

confluent_kafka/avro/serializer/message_serializer.py

@@ -213,4 +249,7 @@ def decode_message(self, message):
            if magic != MAGIC_BYTE:
                raise SerializerError("message does not start with magic byte")
            decoder_func = self._get_decoder_func(schema_id, payload)
-            return decoder_func(payload)
+            return _wrap(


@saabeilin Wouldn't it be better that the _wrap call is moved inside the decoder closure of _get_decoder_func?
Then you don't have to do the self.registry_client.get_by_id(schema_id) calls on each record because the schema is already available at construction time of the decoder function inside _get_decoder_func.

rnpridgeon · 2019-01-30T21:11:37Z

Generic Serde support will provide a means to bind your schema with your data. This is similar to java's 'GenericRecord' implementation. #502

saabeilin added 6 commits September 19, 2018 10:13

Add .idea and .venv to .gitignore

d4e6118

Add Python 3.7 to Tox

6b6dca8

Wrap decoded Avro message and add the schema used for decoding

c0cb293

Create key/value class on the fly correspoding to one provided in Avr…

c874c1c

…o schema

Merge branch 'master' of github.com:confluentinc/confluent-kafka-pyth…

2481ead

…on into avro-consumer-return-schema

More flexible schema-derived class naming.

c339e88

`avro` and `avro-python3` are not 100% compatible. For instance, `PrimitiveSchema` class in python3 has the `name` property while the python2-version does not.

Add tests for schema mixin

68377a0

edenhill suggested changes Sep 25, 2018

View reviewed changes

Address review comments, add docstrings and rename klass

d135353

saabeilin added a commit to saabeilin/kafkian that referenced this pull request Oct 15, 2018

Backport schema mixin.

b9eb7ef

See confluentinc/confluent-kafka-python#453

saabeilin mentioned this pull request Oct 15, 2018

Backport schema mixin saabeilin/kafkian#6

Merged

soxofaan reviewed Oct 26, 2018

View reviewed changes

soxofaan mentioned this pull request Oct 26, 2018

Issue #449: AvroConsumer: provide method to get schemas from last message decode #475

Closed

rnpridgeon closed this Jan 30, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Avro consumer return schema #453

Avro consumer return schema #453

saabeilin commented Sep 20, 2018

ghost commented Sep 20, 2018

rnpridgeon commented Sep 20, 2018

saabeilin commented Sep 20, 2018

rnpridgeon commented Sep 21, 2018

saabeilin commented Sep 24, 2018

edenhill Sep 25, 2018

edenhill Sep 25, 2018

saabeilin Sep 25, 2018

edenhill Sep 25, 2018

edenhill Sep 25, 2018

saabeilin Sep 25, 2018

edenhill Sep 25, 2018

edenhill Sep 25, 2018

saabeilin Sep 25, 2018

saabeilin commented Oct 15, 2018

rnpridgeon commented Oct 15, 2018

soxofaan Oct 26, 2018 •

edited

rnpridgeon commented Jan 30, 2019

Avro consumer return schema #453

Avro consumer return schema #453

Conversation

saabeilin commented Sep 20, 2018

ghost commented Sep 20, 2018

rnpridgeon commented Sep 20, 2018

saabeilin commented Sep 20, 2018

rnpridgeon commented Sep 21, 2018

saabeilin commented Sep 24, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

saabeilin commented Oct 15, 2018

rnpridgeon commented Oct 15, 2018

soxofaan Oct 26, 2018 • edited

Choose a reason for hiding this comment

rnpridgeon commented Jan 30, 2019

soxofaan Oct 26, 2018 •

edited