AvroConsumer for handling schema registry #80

roopahc · 2016-11-22T00:21:10Z

Fixes #36

edenhill · 2016-11-22T15:05:19Z

README.md

+```
+from confluent_kafka.avro import AvroConsumer
+
+c = AvroConsumer({'bootstrap.servers': 'mybroker,mybroker2', 'group.id': 'greoupid', 'schema.registry.url': 'http://127.0.0.1:9002'})


typo: greoupid

the schema registry url port almost looks like 9092 (kafka port), while the default port is 8081, so might want to use that instead

edenhill · 2016-11-22T15:06:50Z

README.md

+    msg = c.poll(10)
+    if msg:
+        if not msg.error():
+            print(msg.value())


Do we want an example that actually uses the deserialized type? This looks just like a standard non-avro consumer.

@edenhill How do we show the type of msg? msg value is going to be a dict object.

No idea, but it is boring if the example is no different than the vanilla example, right? :)

edenhill · 2016-11-22T15:08:43Z

confluent_kafka/avro/__init__.py

+        message = super(AvroConsumer, self).poll(timeout)
+        if not message:
+            return message
+        print(message)


remove print

edenhill · 2016-11-22T15:10:40Z

confluent_kafka/src/confluent_kafka.c

@@ -332,6 +332,19 @@ static PyObject *Message_timestamp (Message *self, PyObject *ignore) {
 			     self->timestamp);
 }

+static PyObject *Message_set_value (Message *self, PyObject *new_val) {
+   Py_DECREF(self->value);


I believe both value and key can be NULL, do:

if (self->value) Py_DECREF(self->value); ``` and same for key

edenhill · 2016-11-22T15:11:24Z

confluent_kafka/src/confluent_kafka.c

@@ -391,6 +404,16 @@ static PyMethodDef Message_methods[] = {
 	  "  :rtype: (int, int)\n"
 	  "\n"
 	},
+	{ "set_value", (PyCFunction)Message_set_value, METH_O,
+	  "  :returns: None.\n"


Missing docstring saying what the method actually does

edenhill · 2016-11-22T18:07:15Z

README.md

 c.subscribe(['my_topic'])
 running = True
 while running:
    msg = c.poll(10)
    if msg:
        if not msg.error():
+            if (msg.value() && isinstance(msg.value(), dict)):


Is this ever not true?
From what I understand the deserializer will raise an exception if it can't deserialize a message.
Speaking of which, this example should probably have a proper try:..except block catching such errors, right?

edenhill · 2016-11-22T18:08:03Z

confluent_kafka/src/confluent_kafka.c

-   Py_INCREF(self->value);
+   if (self->value) {
+        Py_DECREF(self->value);
+        self->value = new_val;


This only sets a new value if there was an old one, which isnt right.
Look at my initial comment.

edenhill · 2016-11-22T18:09:50Z

confluent_kafka/src/confluent_kafka.c

@@ -405,11 +409,13 @@ static PyMethodDef Message_methods[] = {
 	  "\n"
 	},
 	{ "set_value", (PyCFunction)Message_set_value, METH_O,
+	  "  Set the field 'Message.value' with new value.\n"


Document param and type, see an example here:
https://github.com/confluentinc/confluent-kafka-python/blob/master/confluent_kafka/src/confluent_kafka.c#L741

edenhill · 2016-11-22T18:35:44Z

README.md

+            elif msg.error().code() != KafkaError._PARTITION_EOF:
+                print(msg.error())
+                running = False
+    except SerializerError:


where is SerializerError defined? I can't seem to find it

edenhill · 2016-11-22T18:36:08Z

README.md

+                print(msg.error())
+                running = False
+    except SerializerError:
+        print("Unable to decode the message")


print the exception message

edenhill · 2016-11-22T18:37:32Z

confluent_kafka/src/confluent_kafka.c

@@ -410,12 +410,14 @@ static PyMethodDef Message_methods[] = {
 	},
 	{ "set_value", (PyCFunction)Message_set_value, METH_O,
 	  "  Set the field 'Message.value' with new value.\n"
+	  "  :param: PyObject value: Message.value.\n"


PyObject is a C thing, right? Maybe you 'object' or 'anyobject' or such

ewencp · 2016-11-22T22:50:00Z

confluent_kafka/avro/__init__.py

+        if not message:
+            return message
+        if not message.error():
+            if message.value():


I think you want to check if message.value() is not None here. An empty bytes can be a valid message but wouldn't be deserialized here. (In particular this can be an issue if the serialization format does not encode fields that contain their default values, resulting in an empty, but non-null, value.) Same for the check of the key below.

roopahc · 2016-11-23T08:16:46Z

@edenhill @ewencp Let me know if it looks OK now.

edenhill · 2016-11-23T17:50:00Z

README.md

-    except SerializerError:
-        print("Unable to decode the message")
+    except SerializerError as e:
+        exc_type, exc_value, exc_traceback = sys.exc_info()


This is nice, but maybe a bit excessive for an example, I just ment to print the exception string, something like:
print("Message deserialization failed for %s: %s" % (msg, e))

edenhill · 2016-11-27T09:03:20Z

I'm happy with this. @ewencp ?

ewencp · 2016-11-28T18:22:54Z

LGTM

…nt-kafka-python into schema_registry_producer

edenhill · 2016-11-28T20:08:55Z

Thanks again, great work!

roopahc mentioned this pull request Nov 22, 2016

Schema registry client : Consumer code #79

Closed

AvroProducer for handling schema registry

550688d

roopahc force-pushed the schema_registry_producer branch from 926e55e to 550688d Compare November 22, 2016 00:41

edenhill suggested changes Nov 22, 2016

View reviewed changes

Implemented revie feedback

5b88a9e

edenhill suggested changes Nov 22, 2016

View reviewed changes

Feedback

f8ec601

edenhill suggested changes Nov 22, 2016

View reviewed changes

ewencp reviewed Nov 22, 2016

View reviewed changes

Roopa Hiremath Chandrasekaraiah added 2 commits November 22, 2016 19:59

Implemented review comments

4bac7cd

Implemented review comments

21c0b64

edenhill reviewed Nov 23, 2016

View reviewed changes

Implemented review comments

badd3bb

edenhill approved these changes Nov 27, 2016

View reviewed changes

Roopa Hiremath Chandrasekaraiah added 2 commits November 28, 2016 11:40

Consumer client for handling avro schemas

9797fef

Merge branch 'schema_registry_producer' of github.com:roopahc/conflue…

95a9dbb

…nt-kafka-python into schema_registry_producer

edenhill merged commit a9601c4 into confluentinc:master Nov 28, 2016

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AvroConsumer for handling schema registry #80

AvroConsumer for handling schema registry #80

roopahc commented Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

roopahc Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

edenhill Nov 22, 2016

ewencp Nov 22, 2016

roopahc commented Nov 23, 2016

edenhill Nov 23, 2016

edenhill commented Nov 27, 2016

ewencp commented Nov 28, 2016

edenhill commented Nov 28, 2016

AvroConsumer for handling schema registry #80

AvroConsumer for handling schema registry #80

Conversation

roopahc commented Nov 22, 2016

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

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

roopahc commented Nov 23, 2016

Choose a reason for hiding this comment

edenhill commented Nov 27, 2016

ewencp commented Nov 28, 2016

edenhill commented Nov 28, 2016