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

* Enhance error handling for CSFLE configure method

* Add test

* Minor cleanup

* Minor fix

Author: rayokota

schemaregistry/config.go

@@ -88,3 +88,8 @@ func NewConfigWithBearerAuthentication(url, token, targetSr, identityPoolID stri
 
 	return c
 }
+
+// ConfigsEqual compares two configurations for approximate equality
+func ConfigsEqual(c1 *Config, c2 *Config) bool {
+	return internal.ConfigsEqual(&c1.ClientConfig, &c2.ClientConfig)
+}

schemaregistry/internal/client_config.go

@@ -74,3 +74,21 @@ type ClientConfig struct {
 	// HTTP client
 	HTTPClient *http.Client
 }
+
+// ConfigsEqual compares two configurations for approximate equality
+func ConfigsEqual(c1 *ClientConfig, c2 *ClientConfig) bool {
+	return c1.SchemaRegistryURL == c2.SchemaRegistryURL &&
+		c1.BasicAuthUserInfo == c2.BasicAuthUserInfo &&
+		c1.BasicAuthCredentialsSource == c2.BasicAuthCredentialsSource &&
+		c1.SaslMechanism == c2.SaslMechanism &&
+		c1.SaslUsername == c2.SaslUsername &&
+		c1.SaslPassword == c2.SaslPassword &&
+		c1.BearerAuthToken == c2.BearerAuthToken &&
+		c1.BearerAuthCredentialsSource == c2.BearerAuthCredentialsSource &&
+		c1.BearerAuthLogicalCluster == c2.BearerAuthLogicalCluster &&
+		c1.BearerAuthIdentityPoolID == c2.BearerAuthIdentityPoolID &&
+		c1.SslCertificateLocation == c2.SslCertificateLocation &&
+		c1.SslKeyLocation == c2.SslKeyLocation &&
+		c1.SslCaLocation == c2.SslCaLocation &&
+		c1.SslDisableEndpointVerification == c2.SslDisableEndpointVerification
+}

schemaregistry/rules/encryption/encrypt_executor.go

@@ -110,12 +110,32 @@ type FieldEncryptionExecutor struct {
 
 // Configure configures the executor
 func (f *FieldEncryptionExecutor) Configure(clientConfig *schemaregistry.Config, config map[string]string) error {
-	client, err := deks.NewClient(clientConfig)
-	if err != nil {
-		return err
+	if f.Client != nil {
+		if !schemaregistry.ConfigsEqual(f.Client.Config(), clientConfig) {
+			return errors.New("executor already configured")
+		}
+	} else {
+		client, err := deks.NewClient(clientConfig)
+		if err != nil {
+			return err
+		}
+		f.Client = client
+	}
+
+	if f.Config != nil {
+		for key, value := range config {
+			v, ok := f.Config[key]
+			if ok {
+				if v != value {
+					return fmt.Errorf("rule config key already set: %s", key)
+				}
+			} else {
+				f.Config[key] = value
+			}
+		}
+	} else {
+		f.Config = config
 	}
-	f.Config = config
-	f.Client = client
 	return nil
 }
 

schemaregistry/rules/encryption/encrypt_executor_test.go

@@ -0,0 +1,86 @@
+/**
+ * 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.
+ */
+
+package encryption
+
+import (
+	"fmt"
+	"reflect"
+	"runtime"
+	"testing"
+
+	"github.com/confluentinc/confluent-kafka-go/v2/schemaregistry"
+)
+
+func TestFieldEncryptionExecutor_Configure(t *testing.T) {
+	maybeFail = initFailFunc(t)
+
+	executor := NewExecutor()
+	clientConfig := schemaregistry.NewConfig("mock://")
+	config := map[string]string{
+		"key": "value",
+	}
+	err := executor.Configure(clientConfig, config)
+	maybeFail(err)
+	// configure with same args is fine
+	err = executor.Configure(clientConfig, config)
+	maybeFail(err)
+	config2 := map[string]string{
+		"key2": "value2",
+	}
+	// configure with additional config keys is fine
+	err = executor.Configure(clientConfig, config2)
+	maybeFail(err)
+
+	clientConfig2 := schemaregistry.NewConfig("mock://")
+	clientConfig2.BasicAuthUserInfo = "foo"
+	err = executor.Configure(clientConfig2, config)
+	maybeFail(expect(err != nil, true))
+
+	config3 := map[string]string{
+		"key": "value2",
+	}
+	err = executor.Configure(clientConfig, config3)
+	maybeFail(expect(err != nil, true))
+}
+
+type failFunc func(...error)
+
+var maybeFail failFunc
+
+func initFailFunc(t *testing.T) failFunc {
+	tester := t
+	return func(errors ...error) {
+		for _, err := range errors {
+			if err != nil {
+				pc := make([]uintptr, 1)
+				runtime.Callers(2, pc)
+				caller := runtime.FuncForPC(pc[0])
+				_, line := caller.FileLine(caller.Entry())
+
+				tester.Fatalf("%s:%d failed: %s", caller.Name(), line, err)
+			}
+		}
+	}
+}
+
+func expect(actual, expected interface{}) error {
+	if !reflect.DeepEqual(actual, expected) {
+		return fmt.Errorf("expected: %v, Actual: %v", expected, actual)
+	}
+
+	return nil
+}