changefeedccl: support camelCase parameter names for Azure Event Hub sink configuration

Added support for camelCase parameter names (e.g., "SharedAccessKeyName") in Azure
Event Hub Kafka sink configuration while maintaining backward compatibility with
the existing snake_case format (e.g., "shared_access_key_name"). This change
makes the parameter naming more flexible and aligned with Azure's documentation
conventions.

Fixes: #123494
Epic: https://cockroachlabs.atlassian.net/browse/CRDB-38378
Release note (sql change): Added support for camelCase parameter names
(e.g., "SharedAccessKeyName") in Azure Event Hub Kafka sink
configuration

Author: massimo-ua

pkg/ccl/changefeedccl/changefeed.go

@@ -116,6 +116,7 @@ func init() {
 				changefeedbase.SinkParamClientCert,
 				changefeedbase.SinkParamConfluentAPISecret,
 				changefeedbase.SinkParamAzureAccessKey,
+				changefeedbase.SinkParamAzureAccessKeyCamel,
 			})
 			if err != nil {
 				return nil, err

pkg/ccl/changefeedccl/changefeed_test.go

@@ -7046,6 +7046,16 @@ func TestChangefeedErrors(t *testing.T) {
 		`CREATE CHANGEFEED FOR foo INTO $1 WITH topic_in_value, format='experimental_avro'`,
 		`kafka://nope`,
 	)
+	sqlDB.ExpectErrWithTimeout(
+		t, `scheme azure-event-hub cannot specify both shared_access_key and SharedAccessKey`,
+		`CREATE CHANGEFEED FOR foo INTO $1`,
+		`azure-event-hub://nope:9999?SharedAccessKey=redacted&shared_access_key=redacted&shared_access_key_name=saspolicyhistory`,
+	)
+	sqlDB.ExpectErrWithTimeout(
+		t, `scheme azure-event-hub cannot specify both shared_access_key_name and SharedAccessKeyName`,
+		`CREATE CHANGEFEED FOR foo INTO $1`,
+		`azure-event-hub://nope:9999?SharedAccessKeyName=saspolicyhistory&shared_access_key_name=saspolicyhistory&SharedAccessKey=redacted`,
+	)
 
 	// Unordered flag required for some options, disallowed for others.
 	sqlDB.ExpectErrWithTimeout(t, `resolved timestamps cannot be guaranteed to be correct in unordered mode`, `CREATE CHANGEFEED FOR foo WITH resolved, unordered`)

pkg/ccl/changefeedccl/changefeedbase/options.go

@@ -251,9 +251,11 @@ const (
 	SinkParamConfluentAPIKey    = `api_key`
 	SinkParamConfluentAPISecret = `api_secret`
 
-	SinkSchemeAzureKafka        = `azure-event-hub`
-	SinkParamAzureAccessKeyName = `shared_access_key_name`
-	SinkParamAzureAccessKey     = `shared_access_key`
+	SinkSchemeAzureKafka             = `azure-event-hub`
+	SinkParamAzureAccessKeyName      = `shared_access_key_name`
+	SinkParamAzureAccessKeyNameCamel = `SharedAccessKeyName`
+	SinkParamAzureAccessKey          = `shared_access_key`
+	SinkParamAzureAccessKeyCamel     = `SharedAccessKey`
 
 	RegistryParamCACert     = `ca_cert`
 	RegistryParamClientCert = `client_cert`

pkg/ccl/changefeedccl/sink_external_connection.go

@@ -129,6 +129,7 @@ func init() {
 		changefeedbase.SinkParamClientKey,
 		changefeedbase.SinkParamConfluentAPISecret,
 		changefeedbase.SinkParamAzureAccessKey,
+		changefeedbase.SinkParamAzureAccessKeyCamel,
 	))
 }
 

pkg/ccl/changefeedccl/sink_kafka.go

@@ -1065,12 +1065,33 @@ func buildAzureKafkaConfig(u *changefeedbase.SinkURL) (dialConfig kafkaDialConfi
 	hostName := u.Hostname()
 	// saslUser="$ConnectionString"
 	// saslPassword="Endpoint=sb://<NamespaceName>.servicebus.windows.net/;SharedAccessKeyName=<KeyName>;SharedAccessKey=<KeyValue>;
-	sharedAccessKeyName := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKeyName)
+	sharedAccessKeyNameSnake := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKeyName)
+	sharedAccessKeyNameCamel := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKeyNameCamel)
+	if sharedAccessKeyNameSnake != `` && sharedAccessKeyNameCamel != `` {
+		return kafkaDialConfig{}, errors.Newf(`scheme %s cannot specify both %s and %s`, u.Scheme, changefeedbase.SinkParamAzureAccessKeyName, changefeedbase.SinkParamAzureAccessKeyNameCamel)
+	}
+
+	sharedAccessKeyName := sharedAccessKeyNameSnake
+	if sharedAccessKeyName == `` {
+		sharedAccessKeyName = sharedAccessKeyNameCamel
+	}
+
 	if sharedAccessKeyName == `` {
 		return kafkaDialConfig{},
 			newMissingParameterError(u.Scheme /*scheme*/, changefeedbase.SinkParamAzureAccessKeyName /*param*/)
 	}
-	sharedAccessKey := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKey)
+
+	sharedAccessKeySnake := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKey)
+	sharedAccessKeyCamel := u.ConsumeParam(changefeedbase.SinkParamAzureAccessKeyCamel)
+	if sharedAccessKeySnake != `` && sharedAccessKeyCamel != `` {
+		return kafkaDialConfig{}, errors.Newf(`scheme %s cannot specify both %s and %s`, u.Scheme, changefeedbase.SinkParamAzureAccessKey, changefeedbase.SinkParamAzureAccessKeyCamel)
+	}
+
+	sharedAccessKey := sharedAccessKeySnake
+	if sharedAccessKey == `` {
+		sharedAccessKey = sharedAccessKeyCamel
+	}
+
 	if sharedAccessKey == `` {
 		return kafkaDialConfig{},
 			newMissingParameterError(u.Scheme /*scheme*/, changefeedbase.SinkParamAzureAccessKey /*param*/)

pkg/ccl/changefeedccl/sink_kafka_connection_test.go

@@ -471,6 +471,11 @@ func TestAzureKafkaDefaults(t *testing.T) {
 			uri:      "azure-event-hub://myeventhubs.servicebus.windows.net:9093?shared_access_key_name=saspolicyhistory&shared_access_key=q%2BSecretRedacted%3D",
 			expected: makeExpectation("myeventhubs.servicebus.windows.net", "saspolicyhistory", "q+SecretRedacted="),
 		},
+		{
+			name:     "test camel case to snake case param names fallback",
+			uri:      "azure-event-hub://myeventhubs.servicebus.windows.net:9093?SharedAccessKeyName=saspolicyhistory&SharedAccessKey=q%2BSecretRedacted%3D",
+			expected: makeExpectation("myeventhubs.servicebus.windows.net", "saspolicyhistory", "q+SecretRedacted="),
+		},
 	}
 	t.Run("sarama", func(t *testing.T) {
 		for _, tc := range cases {