ProtobufSerializer/ProtobufDeserializer for Generic Serdes API

Author: rnpridgeon

confluent_kafka/schema_registry/protobuf.py

@@ -0,0 +1,5 @@
+user_pb2.py: user.proto
+	protoc -I=. --python_out=. ./user.proto;
+
+clean:
+	rm -f $(TARGET_DIR)/*_pb2.py

examples/Makefile

@@ -0,0 +1,88 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2020 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+#
+# This is a simple example of the SerializingProducer using protobuf.
+#
+# To regenerate Protobuf classes you must first install the protobuf
+# compiler. Once installed you may call protoc directly or use make.
+#
+# See the protocol buffer docs for instructions on installing and using protoc.
+# https://developers.google.com/protocol-buffers/docs/pythontutorial
+#
+# After installing protoc execute the following command from the examples
+# directory to regenerate the user_pb2 module.
+# `make`
+#
+import argparse
+
+# Protobuf generated class; resides at ./user_pb2.py
+import user_pb2
+from confluent_kafka import DeserializingConsumer
+from confluent_kafka.schema_registry.protobuf import ProtobufDeserializer
+from confluent_kafka.serialization import StringDeserializer
+
+
+def main(args):
+    topic = args.topic
+
+    protobuf_deserializer = ProtobufDeserializer(user_pb2.User)
+    string_deserializer = StringDeserializer('utf_8')
+
+    consumer_conf = {'bootstrap.servers': args.bootstrap_servers,
+                     'key.deserializer': string_deserializer,
+                     'value.deserializer': protobuf_deserializer,
+                     'group.id': args.group,
+                     'auto.offset.reset': "earliest"}
+
+    consumer = DeserializingConsumer(consumer_conf)
+    consumer.subscribe([topic])
+
+    while True:
+        try:
+            # SIGINT can't be handled when polling, limit timeout to 1 second.
+            msg = consumer.poll(1.0)
+            if msg is None:
+                continue
+
+            user = msg.value()
+            if user is not None:
+                print("User record {}: name: {}\n"
+                      "\tfavorite_number: {}\n"
+                      "\tfavorite_color: {}\n"
+                      .format(msg.key(), user.name,
+                              user.favorite_color,
+                              user.favorite_number))
+        except KeyboardInterrupt:
+            break
+
+    consumer.close()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description="DeserializingConsumer Example")
+    parser.add_argument('-b', dest="bootstrap_servers", required=True,
+                        help="Bootstrap broker(s) (host[:port])")
+    parser.add_argument('-s', dest="schema_registry", required=True,
+                        help="Schema Registry (http(s)://host[:port]")
+    parser.add_argument('-t', dest="topic", default="example_serde_protobuf",
+                        help="Topic name")
+    parser.add_argument('-g', dest="group", default="example_serde_protobuf",
+                        help="Consumer group")
+
+    main(parser.parse_args())

examples/protobuf_consumer.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2020 Confluent Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+#
+# This is a simple example of the SerializingProducer using protobuf.
+#
+# To regenerate Protobuf classes you must first install the protobuf
+# compiler. Once installed you may call protoc directly or use make.
+#
+# See the protocol buffer docs for instructions on installing and using protoc.
+# https://developers.google.com/protocol-buffers/docs/pythontutorial
+#
+# After installing protoc execute the following command from the examples
+# directory to regenerate the user_pb2 module.
+# `make`
+#
+import argparse
+from uuid import uuid4
+
+from six.moves import input
+
+# Protobuf generated class; resides at ./user_pb2.py
+import user_pb2
+from confluent_kafka import SerializingProducer
+from confluent_kafka.serialization import StringSerializer
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.schema_registry.protobuf import ProtobufSerializer
+
+
+def delivery_report(err, msg):
+    """
+    Reports the failure or success of a message delivery.
+
+    Args:
+        err (KafkaError): The error that occurred on None on success.
+
+        msg (Message): The message that was produced or failed.
+
+    Note:
+        In the delivery report callback the Message.key() and Message.value()
+        will be the binary format as encoded by any configured Serializers and
+        not the same object that was passed to produce().
+        If you wish to pass the original object(s) for key and value to delivery
+        report callback we recommend a bound callback or lambda where you pass
+        the objects along.
+
+    """
+    if err is not None:
+        print("Delivery failed for User record {}: {}".format(msg.key(), err))
+        return
+    print('User record {} successfully produced to {} [{}] at offset {}'.format(
+        msg.key(), msg.topic(), msg.partition(), msg.offset()))
+
+
+def main(args):
+    topic = args.topic
+
+    schema_registry_conf = {'url': args.schema_registry}
+    schema_registry_client = SchemaRegistryClient(schema_registry_conf)
+
+    protobuf_serializer = ProtobufSerializer(user_pb2.User,
+                                             schema_registry_client)
+
+    producer_conf = {'bootstrap.servers': args.bootstrap_servers,
+                     'key.serializer': StringSerializer('utf_8'),
+                     'value.serializer': protobuf_serializer}
+
+    producer = SerializingProducer(producer_conf)
+
+    print("Producing user records to topic {}. ^C to exit.".format(topic))
+    while True:
+        # Serve on_delivery callbacks from previous calls to produce()
+        producer.poll(0.0)
+        try:
+            user_name = input("Enter name: ")
+            user_favorite_number = int(input("Enter favorite number: "))
+            user_favorite_color = input("Enter favorite color: ")
+            user = user_pb2.User(name=user_name,
+                                 favorite_color=user_favorite_color,
+                                 favorite_number=user_favorite_number)
+            producer.produce(topic=topic, key=str(uuid4()), value=user,
+                             on_delivery=delivery_report)
+        except KeyboardInterrupt:
+            break
+        except ValueError:
+            print("Invalid input, discarding record...")
+            continue
+
+    print("\nFlushing records...")
+    producer.flush()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description="SerializingProducer Example")
+    parser.add_argument('-b', dest="bootstrap_servers", required=True,
+                        help="Bootstrap broker(s) (host[:port])")
+    parser.add_argument('-s', dest="schema_registry", required=True,
+                        help="Schema Registry (http(s)://host[:port]")
+    parser.add_argument('-t', dest="topic", default="example_serde_protobuf",
+                        help="Topic name")
+
+    main(parser.parse_args())

examples/protobuf_producer.py

@@ -4,3 +4,4 @@ pydantic
 uvicorn
 six
 jsonschema
+protobuf

examples/requirements.txt

@@ -0,0 +1,7 @@
+syntax = "proto3";
+
+message User {
+    string name = 1;
+    int64 favorite_number = 2;
+    string favorite_color = 3;
+}

examples/user.proto

@@ -0,0 +1,14 @@
+syntax = "proto3";
+
+import "NestedTestProto.proto";
+import "PublicTestProto.proto";
+
+package tests.integration.serialization.data;
+
+option java_package = "io.confluent.kafka.serializers.protobuf.test";
+
+message DependencyMessage {
+    NestedMessage nested_message = 1;
+    bool is_active = 2;
+    TestMessage test_messsage = 3;
+}

examples/user_pb2.py

@@ -13,4 +13,4 @@ compile: $(PROTOS)
 	done
 
 clean:
-	rm $(TARGET_DIR)/*_pb2.py
+	rm -f $(TARGET_DIR)/*_pb2.py

tests/integration/schema_registry/data/DependencyTestProto.proto

@@ -0,0 +1,61 @@
+syntax = "proto3";
+
+import "google/protobuf/timestamp.proto";
+
+package tests.integration.serialization.data;
+
+option java_package = "io.confluent.kafka.serializers.protobuf.test";
+option java_outer_classname = "NestedTestProto";
+option java_multiple_files = false;
+
+message UserId {
+    oneof user_id {
+        string kafka_user_id = 1;
+        int32 other_user_id = 2;
+        MessageId another_id = 3;
+    }
+}
+
+message MessageId {
+    string id = 1;
+}
+
+enum Status {
+    ACTIVE = 0;
+    INACTIVE = 1;
+}
+
+message ComplexType {
+    oneof some_val {
+        string one_id = 1;
+        int32 other_id = 2;
+    }
+    bool is_active = 3;
+}
+
+/*
+ * Complex message using nested protos and repeated fields
+ */
+message NestedMessage {
+    UserId user_id = 1;
+    bool is_active = 2;
+    repeated string experiments_active = 3;
+    Status status = 5;
+    ComplexType complex_type = 6;
+    map<string, string> map_type = 7;
+    InnerMessage inner = 8;
+
+    message InnerMessage {
+        string id = 1 [json_name="id"];
+        repeated int32 ids = 2 [packed=true];
+    }
+
+    enum InnerEnum {
+        option allow_alias = true;
+        ZERO = 0;
+        ALSO_ZERO = 0;
+    }
+
+    reserved 14, 15, 9 to 11;
+    reserved "foo", "bar";
+}