[pkg/kafka] Add validation for AWS MSK SASL config and fix error message

.chloggen/fix-aws-msk-sasl-validation.yaml

@@ -0,0 +1,28 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'bug_fix'
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: 'pkg/kafka/configkafka'
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: 'Fix missing AWS MSK SASL validation and incomplete OAUTHBEARER error message'
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [50934]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Validate that the region is not empty when the AWS MSK mechanism is used, and add the missing OAuth mechanism to the default switch case error message.
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: ['user']

pkg/kafka/configkafka/config.go

@@ -474,7 +474,9 @@ type SASLConfig struct {
 func (c SASLConfig) Validate() error {
 	switch c.Mechanism {
 	case "AWS_MSK_IAM_OAUTHBEARER":
-		// TODO validate c.AWSMSK
+		if c.AWSMSK.Region == "" {
+			return errors.New("region is required for AWS_MSK_IAM_OAUTHBEARER mechanism")
+		}
 	case "PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512":
 		// Do nothing, valid mechanism
 		if c.Username == "" {
@@ -489,7 +491,7 @@ func (c SASLConfig) Validate() error {
 		}
 	default:
 		return fmt.Errorf(
-			"mechanism should be one of 'PLAIN', 'AWS_MSK_IAM_OAUTHBEARER', 'SCRAM-SHA-256' or 'SCRAM-SHA-512'. configured value %v",
+			"mechanism should be one of 'PLAIN', 'AWS_MSK_IAM_OAUTHBEARER', 'OAUTHBEARER', 'SCRAM-SHA-256' or 'SCRAM-SHA-512'. configured value %v",
 			c.Mechanism,
 		)
 	}

pkg/kafka/configkafka/config_test.go

@@ -58,15 +58,7 @@ func TestClientConfig(t *testing.T) {
 				ConnIdleTimeout: 5 * time.Minute,
 			},
 		},
-		"sasl_aws_msk_iam_oauthbearer": {
-			expected: func() ClientConfig {
-				cfg := NewDefaultClientConfig()
-				cfg.Authentication.SASL = &SASLConfig{
-					Mechanism: "AWS_MSK_IAM_OAUTHBEARER",
-				}
-				return cfg
-			}(),
-		},
+
 		"sasl_aws_msk_iam_oauthbearer_with_region": {
 			expected: func() ClientConfig {
 				cfg := NewDefaultClientConfig()
@@ -113,7 +105,10 @@ func TestClientConfig(t *testing.T) {
 			expectedErr: `invalid protocol version: "none"`,
 		},
 		"sasl_invalid_mechanism": {
-			expectedErr: "auth::sasl: mechanism should be one of 'PLAIN', 'AWS_MSK_IAM_OAUTHBEARER', 'SCRAM-SHA-256' or 'SCRAM-SHA-512'. configured value FANCY",
+			expectedErr: "auth::sasl: mechanism should be one of 'PLAIN', 'AWS_MSK_IAM_OAUTHBEARER', 'OAUTHBEARER', 'SCRAM-SHA-256' or 'SCRAM-SHA-512'. configured value FANCY",
+		},
+		"sasl_aws_msk_iam_oauthbearer_missing_region": {
+			expectedErr: "auth::sasl: region is required for AWS_MSK_IAM_OAUTHBEARER mechanism",
 		},
 		"sasl_plain_username_required": {
 			expectedErr: "auth::sasl: username is required",

pkg/kafka/configkafka/testdata/client_config.yaml

@@ -19,10 +19,7 @@ kafka/full:
       max: 10
       backoff: 5s
   conn_idle_timeout: 5m
-kafka/sasl_aws_msk_iam_oauthbearer:
-  auth:
-    sasl:
-      mechanism: AWS_MSK_IAM_OAUTHBEARER
+
 kafka/sasl_aws_msk_iam_oauthbearer_with_region:
   auth:
     sasl:
@@ -59,6 +56,11 @@ kafka/sasl_invalid_mechanism:
     sasl:
       mechanism: FANCY
 
+kafka/sasl_aws_msk_iam_oauthbearer_missing_region:
+  auth:
+    sasl:
+      mechanism: AWS_MSK_IAM_OAUTHBEARER
+
 kafka/sasl_plain_username_required:
   auth:
     sasl: