Support for Data Contracts (#1852)

* First cut at data contracts

* Checkpoint

* Update dependencies

* Add respx

* Fix recursive imports

* Pin httpx

* Minor fix to MeessageToDict call

* Fix flake8 warnings

* Fix flake8 warnings for tests

* Fix npe

* Fix build

* Minor fix

* Add comment

* Add comment

* Minor fix

* Fix tests

* Minor cleanup

* Clean up requirements

* Clean up requirements

* Clean up requirements

* Clean up requirements

* Minor fix

* Add examples

* Minor fix

* Minor fix

* Minor fix

* Minor fix

* Minor fix

* Fix protobuf example

* Handle recursive refs

* Add cache configs

* Add rule overrides

* Minor fix

* Minor fix

* Add override test

* Fix encryption tests

* Minor cleanup

* Add assume role for aws

* Minor fixes

* Incorporate review feedback; consolidate caches

* Add register full response to mock sr

* Fix recursive error during integration test

Author: rayokota

examples/avro/user_generic.avsc

@@ -4,7 +4,8 @@
     "fields": [
         {
             "name": "name",
-            "type": "string"
+            "type": "string",
+            "confluent:tags": ["PII"]
         },
         {
             "name": "favorite_number",

examples/avro/user_specific.avsc

@@ -5,7 +5,8 @@
     "fields": [
         {
             "name": "name",
-            "type": "string"
+            "type": "string",
+            "confluent:tags": ["PII"]
         },
         {
             "name": "favorite_number",

examples/avro_consumer_encryption.py

@@ -0,0 +1,145 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Copyright 2024 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.
+
+
+# A simple example demonstrating use of AvroDeserializer.
+
+import argparse
+
+from confluent_kafka.schema_registry.rules.encryption.encrypt_executor import \
+    FieldEncryptionExecutor
+
+from confluent_kafka.schema_registry.rules.encryption.localkms.local_driver import \
+    LocalKmsDriver
+
+from confluent_kafka.schema_registry.rules.encryption.hcvault.hcvault_driver import \
+    HcVaultKmsDriver
+
+from confluent_kafka.schema_registry.rules.encryption.gcpkms.gcp_driver import \
+    GcpKmsDriver
+
+from confluent_kafka.schema_registry.rules.encryption.azurekms.azure_driver import \
+    AzureKmsDriver
+
+from confluent_kafka.schema_registry.rules.encryption.awskms.aws_driver import \
+    AwsKmsDriver
+
+from confluent_kafka import Consumer
+from confluent_kafka.serialization import SerializationContext, MessageField
+from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.schema_registry.avro import AvroDeserializer
+
+
+class User(object):
+    """
+    User record
+
+    Args:
+        name (str): User's name
+
+        favorite_number (int): User's favorite number
+
+        favorite_color (str): User's favorite color
+    """
+
+    def __init__(self, name=None, favorite_number=None, favorite_color=None):
+        self.name = name
+        self.favorite_number = favorite_number
+        self.favorite_color = favorite_color
+
+
+def dict_to_user(obj, ctx):
+    """
+    Converts object literal(dict) to a User instance.
+
+    Args:
+        obj (dict): Object literal(dict)
+
+        ctx (SerializationContext): Metadata pertaining to the serialization
+            operation.
+    """
+
+    if obj is None:
+        return None
+
+    return User(name=obj['name'],
+                favorite_number=obj['favorite_number'],
+                favorite_color=obj['favorite_color'])
+
+
+def main(args):
+    # Register the KMS drivers and the field-level encryption executor
+    AwsKmsDriver.register()
+    AzureKmsDriver.register()
+    GcpKmsDriver.register()
+    HcVaultKmsDriver.register()
+    LocalKmsDriver.register()
+    FieldEncryptionExecutor.register()
+
+    topic = args.topic
+
+    # When using Data Contract rules, a schema should not be passed to the
+    # AvroDeserializer. The schema is fetched from the Schema Registry.
+    schema_str = None
+
+    sr_conf = {'url': args.schema_registry}
+    schema_registry_client = SchemaRegistryClient(sr_conf)
+
+    avro_deserializer = AvroDeserializer(schema_registry_client,
+                                         schema_str,
+                                         dict_to_user)
+
+    consumer_conf = {'bootstrap.servers': args.bootstrap_servers,
+                     'group.id': args.group,
+                     'auto.offset.reset': "earliest"}
+
+    consumer = Consumer(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 = avro_deserializer(msg.value(), SerializationContext(msg.topic(), MessageField.VALUE))
+            if user is not None:
+                print("User record {}: name: {}\n"
+                      "\tfavorite_number: {}\n"
+                      "\tfavorite_color: {}\n"
+                      .format(msg.key(), user.name,
+                              user.favorite_number,
+                              user.favorite_color))
+        except KeyboardInterrupt:
+            break
+
+    consumer.close()
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description="AvroDeserializer 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_avro",
+                        help="Topic name")
+    parser.add_argument('-g', dest="group", default="example_serde_avro",
+                        help="Consumer group")
+
+    main(parser.parse_args())