DGS-19519 Enhance error handling for CSFLE configure method (#1899)

* Enhance error handling for CSFLE configure method

* Add test

* Minor fix

Author: rayokota

src/confluent_kafka/schema_registry/protobuf.py

@@ -1041,7 +1041,9 @@ def get_type(fd: FieldDescriptor) -> FieldType:
                    FieldDescriptor.TYPE_UINT64, FieldDescriptor.TYPE_FIXED64,
                    FieldDescriptor.TYPE_SFIXED64):
         return FieldType.LONG
-    if fd.type in (FieldDescriptor.TYPE_FLOAT, FieldDescriptor.TYPE_DOUBLE):
+    if fd.type == FieldDescriptor.TYPE_FLOAT:
+        return FieldType.FLOAT
+    if fd.type == FieldDescriptor.TYPE_DOUBLE:
         return FieldType.DOUBLE
     if fd.type == FieldDescriptor.TYPE_BOOL:
         return FieldType.BOOLEAN

src/confluent_kafka/schema_registry/rules/encryption/dek_registry/dek_registry_client.py

@@ -454,6 +454,7 @@ class DekRegistryClient(object):
     """  # noqa: E501
 
     def __init__(self, conf: dict):
+        self._conf = conf
         self._rest_client = _RestClient(conf)
         self._kek_cache = _KekCache()
         self._dek_cache = _DekCache()
@@ -465,6 +466,9 @@ def __exit__(self, *args):
         if self._rest_client is not None:
             self._rest_client.session.close()
 
+    def config(self):
+        return self._conf
+
     def register_kek(
         self, name: str, kms_type: str, kms_key_id: str,
         shared: bool = False, kms_props: Dict[str, str] = None, doc: str = None

src/confluent_kafka/schema_registry/rules/encryption/encrypt_executor.py

@@ -56,8 +56,24 @@ def __init__(self, clock: Clock = Clock()):
         self.clock = clock
 
     def configure(self, client_conf: dict, rule_conf: dict):
-        self.client = DekRegistryClient.new_client(client_conf)
-        self.config = rule_conf
+        if client_conf:
+            if self.client:
+                if self.client.config() != client_conf:
+                    raise RuleError("executor already configured")
+            else:
+                self.client = DekRegistryClient.new_client(client_conf)
+
+        if rule_conf:
+            if self.config:
+                for key, value in rule_conf.items():
+                    v = self.config.get(key)
+                    if v is not None:
+                        if v != value:
+                            raise RuleError(f"rule config key already set: {key}")
+                    else:
+                        self.config[key] = value
+            else:
+                self.config = rule_conf
 
     def type(self) -> str:
         return "ENCRYPT"

tests/schema_registry/test_config.py

@@ -19,6 +19,9 @@
 from httpx import BasicAuth
 
 from confluent_kafka.schema_registry import SchemaRegistryClient
+from confluent_kafka.schema_registry.rules.encryption.encrypt_executor import \
+    FieldEncryptionExecutor
+from confluent_kafka.schema_registry.serde import RuleError
 
 TEST_URL = 'http://SchemaRegistry:65534'
 TEST_USERNAME = 'sr_user'
@@ -119,3 +122,25 @@ def test_config_unknown_prop():
 
     with pytest.raises(ValueError, match=r"Unrecognized properties: (.*)"):
         SchemaRegistryClient(conf)
+
+
+def test_config_encrypt_executor():
+    executor = FieldEncryptionExecutor()
+    client_conf = {'url': 'mock://'}
+    rule_conf = {'key': 'value'}
+    executor.configure(client_conf, rule_conf)
+    # configure with same args is fine
+    executor.configure(client_conf, rule_conf)
+    rule_conf2 = {'key2': 'value2'}
+    # configure with additional rule_conf keys is fine
+    executor.configure(client_conf, rule_conf2)
+
+    client_conf2 = {'url': 'mock://',
+                    'ssl.key.location': '/ssl/keys/client',
+                    'ssl.certificate.location': '/ssl/certs/client'}
+    with pytest.raises(RuleError, match="executor already configured"):
+        executor.configure(client_conf2, rule_conf)
+
+    rule_conf3 = {'key': 'value3'}
+    with pytest.raises(RuleError, match="rule config key already set: key"):
+        executor.configure(client_conf, rule_conf3)