Repository: knative-sandbox/eventing-kafka-broker
Pull request #2161 (2022-05-05 18:08:17)

[release-1.4] Use PLAIN as default SASL if not specified
This is an automated cherry-pick of #2139

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2161#" title="Author self-approved">knative-prow-robot</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **nader-ziada** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @nader-ziada` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.4/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["nader-ziada"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2161](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c5ec1d5) into [release-1.4](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/ebee147bf73edc77145135caa0c845ff7b908223?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ebee147) will **increase** coverage by `0.03%`.
> The diff coverage is `100.00%`.

```diff
@@               Coverage Diff               @@
##           release-1.4    #2161      +/-   ##
===============================================
+ Coverage        61.24%   61.27%   +0.03%     
===============================================
  Files               70       70              
  Lines             6850     6849       -1     
===============================================
+ Hits              4195     4197       +2     
+ Misses            2356     2354       -2     
+ Partials           299      298       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | `100.00% <100.00%> (+2.32%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [ebee147...c5ec1d5](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2161?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow[bot]: @knative-prow-robot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.4 | c5ec1d5ee1efa7ddc1f01fbba5413ae031dc9d2e | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2161/unit-tests_eventing-kafka-broker_release-1.4/1522277019864272896) | true | `/test unit-tests_eventing-kafka-broker_release-1.4`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
pierDipi: /test unit-tests_eventing-kafka-broker_release-1.4

Patch:
diff --git a/control-plane/pkg/security/secret.go b/control-plane/pkg/security/secret.go
index 4dd8c7d82..dcdf67a1e 100644
--- a/control-plane/pkg/security/secret.go
+++ b/control-plane/pkg/security/secret.go
@@ -96,12 +96,12 @@ func secretData(data map[string][]byte) kafka.ConfigOption {
 func saslConfig(protocol string, data map[string][]byte) kafka.ConfigOption {
 	return func(config *sarama.Config) error {
 
-		// Supported mechanism SASL/PLAIN or SASL/SCRAM.
+		// Supported mechanism SASL/PLAIN (default if not specified) or SASL/SCRAM.
+		saslMechanism := SaslPlain
 		givenSASLMechanism, ok := data[SaslMechanismKey]
-		if !ok {
-			return fmt.Errorf("[protocol %s] SASL mechanism required (key: %s)", protocol, SaslMechanismKey)
+		if ok {
+			saslMechanism = string(givenSASLMechanism)
 		}
-		saslMechanism := string(givenSASLMechanism)
 
 		user, ok := data[SaslUserKey]
 		if !ok || len(user) == 0 {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
index bba60df04..5a11923d5 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidator.java
@@ -45,7 +45,7 @@ class CredentialsValidator {
 
     final var SASLMechanism = credentials.SASLMechanism();
     if (is(SecurityProtocol.SASL_PLAINTEXT, securityProtocol)) {
-      if (isInvalidSASLMechanism(SASLMechanism)) {
+      if (SASLMechanism != null && isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256, SCRAM-SHA-512 or PLAIN got " + SASLMechanism;
       }
       if (anyBlank(credentials.SASLUsername(), credentials.SASLPassword())) {
@@ -55,7 +55,7 @@ class CredentialsValidator {
     }
 
     if (is(SecurityProtocol.SASL_SSL, securityProtocol)) {
-      if (isInvalidSASLMechanism(SASLMechanism)) {
+      if (SASLMechanism != null && isInvalidSASLMechanism(SASLMechanism)) {
         return "Security protocol " + securityProtocol.name + ": invalid SASL mechanism, expected SCRAM-SHA-256, SCRAM-SHA-512 or PLAIN got " + SASLMechanism;
       }
       if (anyBlank(credentials.SASLUsername(), credentials.SASLPassword())) {
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
index 8e371cac4..b85d03c2d 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuth.java
@@ -57,9 +57,12 @@ public class KafkaClientsAuth {
   }
 
   private static void sasl(final BiConsumer<String, String> propertiesSetter, final Credentials credentials) {
-    final var mechanism = credentials.SASLMechanism();
-    if (mechanism == null) {
-      throw new IllegalStateException("SASL mechanism required");
+    final String mechanism;
+    final var saslMechanism = credentials.SASLMechanism();
+    if (saslMechanism == null) {
+      mechanism = "PLAIN";
+    } else {
+      mechanism = saslMechanism;
     }
     propertiesSetter.accept(SaslConfigs.SASL_MECHANISM, mechanism);
     if ("PLAIN".equals(mechanism)) {


Test patch:
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
index 75796af45..1e02418ee 100644
--- a/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2_test.go
@@ -21,8 +21,13 @@ import (
 	"fmt"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	clientgotesting "k8s.io/client-go/testing"
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/kmeta"
 
@@ -94,6 +99,244 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal with SASL with type",
+			Objects: []runtime.Object{
+				//NewSourceSASL(),
+				&sources.KafkaSource{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: SourceNamespace,
+						Name:      SourceName,
+						UID:       SourceUUID,
+					},
+					Spec: sources.KafkaSourceSpec{
+						KafkaAuthSpec: v1beta1.KafkaAuthSpec{
+							BootstrapServers: []string{SourceBootstrapServers},
+							Net: v1beta1.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+									Type: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "type",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						},
+						Topics:        SourceTopics,
+						ConsumerGroup: SourceConsumerGroup,
+						SourceSpec: duckv1.SourceSpec{
+							Sink: NewSourceSinkReference(),
+						},
+					},
+				},
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(&kafkainternals.Auth{
+							NetSpec: &bindings.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+									Type: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "type",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						}),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						SourceNetSaslTls(true),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal with SASL without type",
+			Objects: []runtime.Object{
+				//NewSourceSASL(),
+				&sources.KafkaSource{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: SourceNamespace,
+						Name:      SourceName,
+						UID:       SourceUUID,
+					},
+					Spec: sources.KafkaSourceSpec{
+						KafkaAuthSpec: v1beta1.KafkaAuthSpec{
+							BootstrapServers: []string{SourceBootstrapServers},
+							Net: v1beta1.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						},
+						Topics:        SourceTopics,
+						ConsumerGroup: SourceConsumerGroup,
+						SourceSpec: duckv1.SourceSpec{
+							Sink: NewSourceSinkReference(),
+						},
+					},
+				},
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(SourceUUID),
+					WithConsumerGroupNamespace(SourceNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(NewSource())),
+					WithConsumerGroupMetaLabels(OwnerAsSourceLabel),
+					WithConsumerGroupLabels(ConsumerSourceLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics[0], SourceTopics[1]),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+						),
+						ConsumerAuth(&kafkainternals.Auth{
+							NetSpec: &bindings.KafkaNetSpec{
+								SASL: bindings.KafkaSASLSpec{
+									Enable: true,
+									User: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "user",
+										},
+									},
+									Password: bindings.SecretValueFromSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{
+												Name: SecretName,
+											},
+											Key: "password",
+										},
+									},
+								},
+								TLS: bindings.KafkaTLSSpec{
+									Enable: true,
+								},
+							},
+						}),
+						ConsumerDelivery(
+							NewConsumerSpecDelivery(
+								internals.Ordered,
+								NewConsumerTimeout("PT600S"),
+								NewConsumerRetry(10),
+								NewConsumerBackoffDelay("PT10S"),
+								NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+							),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerReply(ConsumerNoReply()),
+					)),
+					ConsumerGroupReplicas(1),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSource(
+						StatusSourceConsumerGroupUnknown(),
+						StatusSourceSinkResolved(""),
+						SourceNetSaslTls(false),
+					),
+				},
+			},
+		},
 		{
 			Name: "Reconciled normal - ce overrides",
 			Objects: []runtime.Object{
@@ -558,3 +801,51 @@ func StatusSourceConsumerGroupReplicas(replicas int32) KRShapedOption {
 		ks.Status.Consumers = replicas
 	}
 }
+
+func SourceNetSaslTls(withType bool) KRShapedOption {
+	return func(obj duckv1.KRShaped) {
+		ks := obj.(*sources.KafkaSource)
+		ks.Spec.Net = bindings.KafkaNetSpec{
+			SASL: bindings.KafkaSASLSpec{
+				Enable: true,
+				User: bindings.SecretValueFromSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: SecretName,
+						},
+						Key: "user",
+					},
+				},
+				Password: bindings.SecretValueFromSource{
+					SecretKeyRef: &corev1.SecretKeySelector{
+						LocalObjectReference: corev1.LocalObjectReference{
+							Name: SecretName,
+						},
+						Key: "password",
+					},
+				},
+				// Type: bindings.SecretValueFromSource{
+				// 	SecretKeyRef: &corev1.SecretKeySelector{
+				// 		LocalObjectReference: corev1.LocalObjectReference{
+				// 			Name: SecretName,
+				// 		},
+				// 		Key: "type",
+				// 	},
+				// },
+			},
+			TLS: bindings.KafkaTLSSpec{
+				Enable: true,
+			},
+		}
+		if withType {
+			ks.Spec.Net.SASL.Type = bindings.SecretValueFromSource{
+				SecretKeyRef: &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{
+						Name: SecretName,
+					},
+					Key: "type",
+				},
+			}
+		}
+	}
+}
diff --git a/control-plane/pkg/security/secret_test.go b/control-plane/pkg/security/secret_test.go
index 604702f77..04305c539 100644
--- a/control-plane/pkg/security/secret_test.go
+++ b/control-plane/pkg/security/secret_test.go
@@ -75,6 +75,24 @@ func TestSASLPlain(t *testing.T) {
 	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
 }
 
+func TestSASLPlainDefaulted(t *testing.T) {
+	secret := map[string][]byte{
+		"protocol": []byte("SASL_PLAINTEXT"),
+		"user":     []byte("my-user-name"),
+		"password": []byte("my-user-password"),
+	}
+	config := sarama.NewConfig()
+
+	err := kafka.Options(config, secretData(secret))
+
+	assert.Nil(t, err)
+	assert.True(t, config.Net.SASL.Enable)
+	assert.True(t, config.Net.SASL.Handshake)
+	assert.Equal(t, sarama.SASLMechanism(sarama.SASLTypePlaintext), config.Net.SASL.Mechanism)
+	assert.Equal(t, "my-user-name", config.Net.SASL.User)
+	assert.Equal(t, "my-user-password", config.Net.SASL.Password)
+}
+
 func TestSASLPlainLSCRAM256(t *testing.T) {
 	secret := map[string][]byte{
 		"protocol":       []byte("SASL_PLAINTEXT"),
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
index 49d590128..fcfd39e9e 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsValidatorTest.java
@@ -110,12 +110,19 @@ public class CredentialsValidatorTest {
     securityProtocolSaslPlaintextValid("PLAIN");
   }
 
+  @Test
+  public void securityProtocolSaslPlaintextDefauledPlainValid() {
+    securityProtocolSaslPlaintextValid(null);
+  }
+
   private static void securityProtocolSaslPlaintextValid(final String mechanism) {
 
     final var credential = mock(Credentials.class);
 
     when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_PLAINTEXT);
-    when(credential.SASLMechanism()).thenReturn(mechanism);
+    if(mechanism != null) {
+      when(credential.SASLMechanism()).thenReturn(mechanism);
+    }
     when(credential.SASLUsername()).thenReturn("aaa");
     when(credential.SASLPassword()).thenReturn("bbb");
 
@@ -184,6 +191,18 @@ public class CredentialsValidatorTest {
     assertThat(CredentialsValidator.validate(credential)).isNull();
   }
 
+  @Test
+  public void securityProtocolSaslDefaultedPlainSslValid() {
+    final var credential = mock(Credentials.class);
+
+    when(credential.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
+    when(credential.caCertificates()).thenReturn("xyz");
+    when(credential.SASLUsername()).thenReturn("aaa");
+    when(credential.SASLPassword()).thenReturn("bbb");
+
+    assertThat(CredentialsValidator.validate(credential)).isNull();
+  }
+
   @Test
   public void securityProtocolSaslSslScramSha513InValid() {
     final var credential = mock(Credentials.class);
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
index 8b08e44f5..50ad2d433 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KafkaClientsAuthTest.java
@@ -49,13 +49,20 @@ public class KafkaClientsAuthTest {
     shouldConfigureSaslSsl(PlainLoginModule.class, "PLAIN");
   }
 
+  @Test
+  public void shouldConfigureSaslDefaultedPlainSsl() {
+    shouldConfigureSaslSsl(PlainLoginModule.class, null);
+  }
+
   private static void shouldConfigureSaslSsl(final Class<? extends LoginModule> module, final String mechanism) {
     final var props = new Properties();
 
     final var credentials = mock(Credentials.class);
     when(credentials.securityProtocol()).thenReturn(SecurityProtocol.SASL_SSL);
     when(credentials.caCertificates()).thenReturn("xyz");
-    when(credentials.SASLMechanism()).thenReturn(mechanism);
+    if(mechanism != null) {
+      when(credentials.SASLMechanism()).thenReturn(mechanism);
+    }
     when(credentials.SASLUsername()).thenReturn("aaa");
     when(credentials.SASLPassword()).thenReturn("bbb");
 
@@ -65,7 +72,12 @@ public class KafkaClientsAuthTest {
     expected.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SASL_SSL.name());
     expected.setProperty(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG, DefaultSslEngineFactory.PEM_TYPE);
     expected.setProperty(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG, "xyz");
-    expected.setProperty(SaslConfigs.SASL_MECHANISM, mechanism);
+    if(mechanism != null) {
+      expected.setProperty(SaslConfigs.SASL_MECHANISM, mechanism);
+    } else {
+      expected.setProperty(SaslConfigs.SASL_MECHANISM, "PLAIN"); // it is defaulted
+
+    }
     expected.setProperty(
       SaslConfigs.SASL_JAAS_CONFIG,
       module.getName() + " required username=\"" + credentials.SASLUsername() + "\" password=\"" +