Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Migration to new KafkaChannel does take into account authentication defaults from old implementation
**Describe the bug**
The previous implementation of the channel did not require `protocol` and `sasl.mechanism` to be present in the Secret that was used for authentication.
The `protocol` settings is not required otherwise the Broker will fail with this error:
```
reason: 'error getting cluster admin config: protocol required (key: protocol)
      supported protocols: [PLAINTEXT SASL_PLAINTEXT SSL SASL_SSL]'
```
The sasl.mechanism now defaults to PLAIN as implemented in https://github.com/knative-sandbox/eventing-kafka-broker/issues/2117 but if the previous implementation was using some other mechanism, the new default (PLAIN) will not work with the existing Kafka cluster. This effectively breaks the existing deployments and the KafkaChannel stops working during the upgrade.

 **Expected behavior**
The protocol and sasl.mechanism is properly migrated when migrating from old to new channel implementation.

 **To Reproduce**
Assuming a Strimzi-installed Kafka in kafka namespace with:
```
       - name: sasl
         port: 9094
         type: internal
         tls: true
         authentication:
           type: scram-sha-512
```
and a KafkaUser my-sasl-user user.

* In a knative-eventing namespace, create the secret that will be used by KnativeKafka.
```
STRIMZI_CRT=$(oc -n kafka get secret my-cluster-cluster-ca-cert --template='{{index .data "ca.crt"}}' | base64 --decode )
SASL_PASSWD=$(oc -n kafka get secret my-sasl-user --template='{{index .data "password"}}' | base64 --decode )
oc create secret --namespace knative-eventing generic my-sasl-secret \
      --from-literal=ca.crt="$STRIMZI_CRT" \
      --from-literal=password="$SASL_PASSWD" \
      --from-literal=saslType="SCRAM-SHA-512" \
      --from-literal=user="my-sasl-user"
```
The secret above works fine with the previous implementation of the Channel.

* Now install the previous version of KnativeEventing and KnativeKafka with the following configs:
kafka-channel-config:
```
kind: ConfigMap
apiVersion: v1
metadata:
  name: kafka-channel-config
  namespace: knative-eventing
data:
  auth.secret.ref.name: my-sasl-secret
  auth.secret.ref.namespace: knative-eventing
  bootstrap.servers: 'my-cluster-kafka-bootstrap.kafka:9094'
```
config-br-default-channel:
```
kind: ConfigMap
apiVersion: v1
metadata:
  name: config-br-default-channel
  namespace: knative-eventing
data:
  channel-template-spec: |-
    apiVersion: messaging.knative.dev/v1beta1
    kind: KafkaChannel
    spec:
      numPartitions: 6
      replicationFactor: 3
      delivery:
        retry: 12
        backoffPolicy: exponential
        backoffDelay: PT1S
```
config-br-defaults:
```
kind: ConfigMap
apiVersion: v1
metadata:
  name: config-br-defaults
  namespace: knative-eventing
data:
  default-br-config: |
    clusterDefault:
      apiVersion: v1
      brokerClass: MTChannelBasedBroker
      delivery:
        backoffDelay: PT0.2S
        backoffPolicy: exponential
        retry: 10
      kind: ConfigMap
      name: config-br-default-channel
      namespace: knative-eventing
```
* Create a channel like this:
```
apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: my-broker
  namespace: my-namespace
```
* Upgrade Eventing to the latest version which uses the new implementation.
* The channel ends up with error like this:
```
apiVersion: messaging.knative.dev/v1beta1
kind: KafkaChannel
metadata:
  annotations:
    eventing.knative.dev/scope: cluster
  creationTimestamp: "2022-05-23T08:47:23Z"
  deletionGracePeriodSeconds: 0
  deletionTimestamp: "2022-05-23T08:57:39Z"
  finalizers:
  - kafkachannels.messaging.knative.dev
  generation: 2
  labels:
    eventing.knative.dev/broker: my-broker
    eventing.knative.dev/brokerEverything: "true"
  name: my-broker-kne-trigger
  namespace: serverless-tests
  ownerReferences:
  - apiVersion: eventing.knative.dev/v1
    blockOwnerDeletion: true
    controller: true
    kind: Broker
    name: my-broker
    uid: ef34b9f7-acc7-44bd-affb-31e42d5fc2c5
  resourceVersion: "125460"
  uid: 966c5534-2fb1-4e6b-9279-8f0fc402dfad
spec:
  delivery:
    backoffDelay: PT1S
    backoffPolicy: exponential
    retry: 12
  numPartitions: 6
  replicationFactor: 3
status:
  conditions:
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    status: Unknown
    type: Addressable
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    status: Unknown
    type: ConfigMapUpdated
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    status: "True"
    type: ConfigParsed
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    status: "True"
    type: DataPlaneAvailable
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    status: Unknown
    type: ProbeSucceeded
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    message: 'error getting cluster admin sarama config: protocol required (key: protocol)
      supported protocols: [PLAINTEXT SASL_PLAINTEXT SSL SASL_SSL]'
    reason: 'Failed to create topic: knative-messaging-kafka.serverless-tests.my-broker-kne-trigger'
    status: "False"
    type: Ready
  - lastTransitionTime: "2022-05-23T08:47:23Z"
    message: 'error getting cluster admin sarama config: protocol required (key: protocol)
      supported protocols: [PLAINTEXT SASL_PLAINTEXT SSL SASL_SSL]'
    reason: 'Failed to create topic: knative-messaging-kafka.serverless-tests.my-broker-kne-trigger'
    status: "False"
    type: TopicReady
  observedGeneration: 1
```

Note: If the secret was created like below the new implementation works as expected after upgrade:
```
oc create secret --namespace knative-eventing generic my-sasl-secret \
      --from-literal=ca.crt="$STRIMZI_CRT" \
      --from-literal=password="$SASL_PASSWD" \
      --from-literal=saslType="SCRAM-SHA-512" \
      --from-literal=user="my-sasl-user" \
      --from-literal=protocol="SASL_SSL" \  #This is added.
      --from-literal=sasl.mechanism="SCRAM-SHA-512" # This is added.
```


 **Knative release version**
1.2.3



Pull request:
Support legacy KafkaChannel secret format
Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

Fixes #2213

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 37e868f41..33104e4d4 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -147,8 +147,13 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 		)
 	}
 
+	authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
+	if err != nil {
+		return statusConditionManager.FailedToResolveConfig(fmt.Errorf("failed to resolve auth context: %w", err))
+	}
+
 	// get security option for Sarama with secret info in it
-	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(authContext.VirtualSecret)
 
 	if err := r.TrackSecret(secret, channel); err != nil {
 		return fmt.Errorf("failed to track secret: %w", err)
@@ -204,7 +209,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	// Get resource configuration
-	channelResource, err := r.getChannelContractResource(ctx, topic, channel, secret, topicConfig)
+	channelResource, err := r.getChannelContractResource(ctx, topic, channel, authContext, topicConfig)
 	if err != nil {
 		return statusConditionManager.FailedToResolveConfig(err)
 	}
@@ -402,8 +407,13 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 		)
 	}
 
+	authContext, err := security.ResolveAuthContextFromLegacySecret(secret)
+	if err != nil {
+		return fmt.Errorf("failed to resolve auth context: %w", err)
+	}
+
 	// get security option for Sarama with secret info in it
-	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+	saramaSecurityOption := security.NewSaramaSecurityOptionFromSecret(authContext.VirtualSecret)
 
 	saramaConfig, err := kafka.GetSaramaConfig(saramaSecurityOption)
 	if err != nil {
@@ -565,7 +575,7 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, cm *corev1.ConfigMap, chann
 	}, nil
 }
 
-func (r *Reconciler) getChannelContractResource(ctx context.Context, topic string, channel *messagingv1beta1.KafkaChannel, secret *corev1.Secret, config *kafka.TopicConfig) (*contract.Resource, error) {
+func (r *Reconciler) getChannelContractResource(ctx context.Context, topic string, channel *messagingv1beta1.KafkaChannel, auth *security.NetSpecAuthContext, config *kafka.TopicConfig) (*contract.Resource, error) {
 	resource := &contract.Resource{
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
@@ -580,14 +590,9 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		},
 	}
 
-	if secret != nil {
-		resource.Auth = &contract.Resource_AuthSecret{
-			AuthSecret: &contract.Reference{
-				Uuid:      string(secret.UID),
-				Namespace: secret.Namespace,
-				Name:      secret.Name,
-				Version:   secret.ResourceVersion,
-			},
+	if auth != nil && auth.MultiSecretReference != nil {
+		resource.Auth = &contract.Resource_MultiAuthSecret{
+			MultiAuthSecret: auth.MultiSecretReference,
 		}
 	}
 
diff --git a/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go b/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go
new file mode 100644
index 000000000..81639ebb7
--- /dev/null
+++ b/control-plane/pkg/security/secrets_provider_legacy_channel_secret.go
@@ -0,0 +1,125 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * 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 security
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"knative.dev/eventing-kafka/pkg/common/client"
+	legacycommonconfig "knative.dev/eventing-kafka/pkg/common/config"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+)
+
+func ResolveAuthContextFromLegacySecret(s *corev1.Secret) (*NetSpecAuthContext, error) {
+	if s == nil {
+		return &NetSpecAuthContext{}, nil
+	}
+
+	legacyAuth := legacycommonconfig.GetAuthConfigFromSecret(s)
+	protocolStr, protocolContract := extractProtocol(legacyAuth)
+
+	virtualSecret := s.DeepCopy()
+	virtualSecret.Data[ProtocolKey] = []byte(protocolStr)
+	if v, ok := virtualSecret.Data["sasltype"]; ok {
+		virtualSecret.Data[SaslMechanismKey] = v
+	}
+	if v, ok := virtualSecret.Data["saslType"]; ok {
+		virtualSecret.Data[SaslMechanismKey] = v
+	}
+	if v, ok := virtualSecret.Data["username"]; ok {
+		virtualSecret.Data[SaslUserKey] = v
+	}
+	if v, ok := virtualSecret.Data["user"]; ok {
+		virtualSecret.Data[SaslUserKey] = v
+	}
+
+	return &NetSpecAuthContext{
+		VirtualSecret: virtualSecret,
+		MultiSecretReference: &contract.MultiSecretReference{
+			Protocol:   protocolContract,
+			References: resolveReferencesFromLegacyKafkaSecret(s),
+		},
+	}, nil
+}
+
+func resolveReferencesFromLegacyKafkaSecret(s *corev1.Secret) []*contract.SecretReference {
+	sRef := &contract.SecretReference{
+		Reference: &contract.Reference{
+			Uuid:      string(s.UID),
+			Namespace: s.GetNamespace(),
+			Name:      s.GetName(),
+			Version:   s.ResourceVersion,
+		},
+	}
+
+	hasFields := maybeAddKeyFieldRef(s, UserKey, contract.SecretField_USER_KEY, sRef)
+	hasFields = maybeAddKeyFieldRef(s, UserCertificate, contract.SecretField_USER_CRT, sRef) || hasFields
+	hasFields = maybeAddKeyFieldRef(s, CaCertificateKey, contract.SecretField_CA_CRT, sRef) || hasFields
+	hasFields = maybeAddKeyFieldRef(s, SaslPasswordKey, contract.SecretField_PASSWORD, sRef) || hasFields
+	// We need to support `username` and `user` since in some cases one is used over the other
+	hasFields = maybeAddKeyFieldRef(s, "user", contract.SecretField_USER, sRef) || hasFields
+	if _, ok := s.Data["user"]; !ok {
+		hasFields = maybeAddKeyFieldRef(s, "username", contract.SecretField_USER, sRef) || hasFields
+	}
+	// We need to support `saslType` and `sasltype` since in some cases one is used over the other
+	hasFields = maybeAddKeyFieldRef(s, "saslType", contract.SecretField_SASL_MECHANISM, sRef) || hasFields
+	if _, ok := s.Data["saslType"]; !ok {
+		hasFields = maybeAddKeyFieldRef(s, "sasltype", contract.SecretField_SASL_MECHANISM, sRef) || hasFields
+	}
+
+	if !hasFields {
+		return nil
+	}
+
+	return []*contract.SecretReference{sRef}
+}
+
+func maybeAddKeyFieldRef(s *corev1.Secret, key string, field contract.SecretField, reference *contract.SecretReference) bool {
+	if s.Data == nil {
+		return false
+	}
+
+	if v, ok := s.Data[key]; ok && len(v) > 0 {
+		reference.KeyFieldReferences = append(reference.KeyFieldReferences, &contract.KeyFieldReference{
+			SecretKey: key,
+			Field:     field,
+		})
+		return true
+	}
+	return false
+}
+
+func extractProtocol(auth *client.KafkaAuthConfig) (string, contract.Protocol) {
+	if hasTLSEnabled(auth) && hasSASLEnabled(auth) {
+		return ProtocolSASLSSL, contract.Protocol_SASL_SSL
+	}
+	if hasTLSEnabled(auth) {
+		return ProtocolSSL, contract.Protocol_SSL
+	}
+	if hasSASLEnabled(auth) {
+		return ProtocolSASLPlaintext, contract.Protocol_SASL_PLAINTEXT
+	}
+	return ProtocolPlaintext, contract.Protocol_PLAINTEXT
+}
+
+func hasTLSEnabled(auth *client.KafkaAuthConfig) bool {
+	return auth.TLS != nil && (auth.TLS.Userkey != "" || auth.TLS.Cacert != "" || auth.TLS.Usercert != "")
+}
+
+func hasSASLEnabled(auth *client.KafkaAuthConfig) bool {
+	return auth.SASL != nil && auth.SASL.User != "" || auth.SASL.Password != ""
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 4028cac26..62cd532f0 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -1062,7 +1062,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 		},
 		{
-			Name: "Reconciled normal - with single fresh subscriber - with auth - PlainText",
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SSL",
 			Objects: []runtime.Object{
 				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
 				NewService(),
@@ -1081,7 +1081,7 @@ func TestReconcileKind(t *testing.T) {
 					security.AuthSecretNamespaceKey:    "ns-1",
 				}),
 				NewConfigMapWithBinaryData(&env, nil),
-				NewSSLSecret("ns-1", "secret-1"),
+				NewLegacySSLSecret("ns-1", "secret-1"),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -1093,12 +1093,213 @@ func TestReconcileKind(t *testing.T) {
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
-							Auth: &contract.Resource_AuthSecret{
-								AuthSecret: &contract.Reference{
-									Uuid:      SecretUUID,
-									Namespace: "ns-1",
-									Name:      "secret-1",
-									Version:   SecretResourceVersion,
+							Auth: &contract.Resource_MultiAuthSecret{
+								MultiAuthSecret: &contract.MultiSecretReference{
+									Protocol: contract.Protocol_SSL,
+									References: []*contract.SecretReference{{
+										Reference: &contract.Reference{
+											Uuid:      SecretUUID,
+											Namespace: "ns-1",
+											Name:      "secret-1",
+											Version:   SecretResourceVersion,
+										},
+										KeyFieldReferences: []*contract.KeyFieldReference{
+											{SecretKey: "user.key", Field: contract.SecretField_USER_KEY},
+											{SecretKey: "user.crt", Field: contract.SecretField_USER_CRT},
+											{SecretKey: "ca.crt", Field: contract.SecretField_CA_CRT},
+										},
+									}},
+								},
+							},
+							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SASL SSL",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
+				NewService(),
+				NewPerChannelService(DefaultEnv),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+					security.AuthSecretNameKey:         "secret-1",
+					security.AuthSecretNamespaceKey:    "ns-1",
+				}),
+				NewConfigMapWithBinaryData(&env, nil),
+				NewLegacySASLSSLSecret("ns-1", "secret-1"),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Auth: &contract.Resource_MultiAuthSecret{
+								MultiAuthSecret: &contract.MultiSecretReference{
+									Protocol: contract.Protocol_SASL_SSL,
+									References: []*contract.SecretReference{{
+										Reference: &contract.Reference{
+											Uuid:      SecretUUID,
+											Namespace: "ns-1",
+											Name:      "secret-1",
+											Version:   SecretResourceVersion,
+										},
+										KeyFieldReferences: []*contract.KeyFieldReference{
+											{SecretKey: "user.key", Field: contract.SecretField_USER_KEY},
+											{SecretKey: "user.crt", Field: contract.SecretField_USER_CRT},
+											{SecretKey: "ca.crt", Field: contract.SecretField_CA_CRT},
+											{SecretKey: "password", Field: contract.SecretField_PASSWORD},
+											{SecretKey: "username", Field: contract.SecretField_USER},
+											{SecretKey: "saslType", Field: contract.SecretField_SASL_MECHANISM},
+										},
+									}},
+								},
+							},
+							Ingress: &contract.Ingress{
+								Path: receiver.Path(ChannelNamespace, ChannelName),
+							},
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
+								Destination:   "http://" + Subscription1URI,
+								Uid:           Subscription1UUID,
+								DeliveryOrder: contract.DeliveryOrder_ORDERED,
+								ReplyStrategy: &contract.Egress_ReplyUrl{ReplyUrl: "http://" + Subscription1ReplyURI},
+							}},
+						},
+					},
+				}),
+				ChannelReceiverPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				ChannelDispatcherPodUpdate(env.SystemNamespace, map[string]string{
+					"annotation_to_preserve":           "value_to_preserve",
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusConfigParsed,
+						StatusConfigMapUpdatedReady(&env),
+						StatusTopicReadyWithName(ChannelTopic()),
+						StatusDataPlaneAvailable,
+						ChannelAddressable(&env),
+						WithSubscribers(Subscriber1()),
+						StatusProbeSucceeded,
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+		},
+		{
+			Name: "Reconciled normal - with single fresh subscriber - with auth - SASL",
+			Objects: []runtime.Object{
+				NewChannel(WithSubscribers(Subscriber1(WithFreshSubscriber))),
+				NewService(),
+				NewPerChannelService(DefaultEnv),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+					security.AuthSecretNameKey:         "secret-1",
+					security.AuthSecretNamespaceKey:    "ns-1",
+				}),
+				NewConfigMapWithBinaryData(&env, nil),
+				NewLegacySASLSecret("ns-1", "secret-1"),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&env, &contract.Contract{
+					Generation: 1,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ChannelUUID,
+							Topics:           []string{ChannelTopic()},
+							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
+							Auth: &contract.Resource_MultiAuthSecret{
+								MultiAuthSecret: &contract.MultiSecretReference{
+									Protocol: contract.Protocol_SASL_PLAINTEXT,
+									References: []*contract.SecretReference{{
+										Reference: &contract.Reference{
+											Uuid:      SecretUUID,
+											Namespace: "ns-1",
+											Name:      "secret-1",
+											Version:   SecretResourceVersion,
+										},
+										KeyFieldReferences: []*contract.KeyFieldReference{
+											{SecretKey: "password", Field: contract.SecretField_PASSWORD},
+											{SecretKey: "username", Field: contract.SecretField_USER},
+											{SecretKey: "saslType", Field: contract.SecretField_SASL_MECHANISM},
+										},
+									}},
 								},
 							},
 							Ingress: &contract.Ingress{
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 6ce05c85f..c803e8288 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -190,6 +190,64 @@ func NewSSLSecret(ns, name string) *corev1.Secret {
 	}
 }
 
+func NewLegacySSLSecret(ns, name string) *corev1.Secret {
+
+	ca, userKey, userCert := loadCerts()
+
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:       ns,
+			Name:            name,
+			ResourceVersion: SecretResourceVersion,
+			UID:             SecretUUID,
+		},
+		Data: map[string][]byte{
+			security.CaCertificateKey: ca,
+			security.UserKey:          userKey,
+			security.UserCertificate:  userCert,
+		},
+	}
+}
+
+func NewLegacySASLSSLSecret(ns, name string) *corev1.Secret {
+
+	ca, userKey, userCert := loadCerts()
+
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:       ns,
+			Name:            name,
+			ResourceVersion: SecretResourceVersion,
+			UID:             SecretUUID,
+		},
+		Data: map[string][]byte{
+			security.CaCertificateKey: ca,
+			security.UserKey:          userKey,
+			security.UserCertificate:  userCert,
+			"username":                []byte("user"),
+			security.SaslPasswordKey:  []byte("password"),
+			"saslType":                []byte("PLAIN"),
+		},
+	}
+}
+
+func NewLegacySASLSecret(ns, name string) *corev1.Secret {
+
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:       ns,
+			Name:            name,
+			ResourceVersion: SecretResourceVersion,
+			UID:             SecretUUID,
+		},
+		Data: map[string][]byte{
+			"username":               []byte("user"),
+			security.SaslPasswordKey: []byte("password"),
+			"saslType":               []byte("PLAIN"),
+		},
+	}
+}
+
 func NewSASLSSLSecret(ns, name string) *corev1.Secret {
 
 	ca, userKey, userCert := loadCerts()
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index 31a6fa4c5..24186eae5 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -348,13 +348,13 @@ function setup_kafka_channel_auth() {
     kubectl patch configmap/kafka-channel-config \
       -n knative-eventing \
       --type merge \
-      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9094", "auth.secret.ref.name": "strimzi-sasl-secret", "auth.secret.ref.namespace": "knative-eventing"}}'
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9094", "auth.secret.ref.name": "strimzi-sasl-secret-legacy", "auth.secret.ref.namespace": "knative-eventing"}}'
   elif [ "$EVENTING_KAFKA_BROKER_CHANNEL_AUTH_SCENARIO" == "SASL_PLAIN" ]; then
     echo "Setting up SASL_PLAIN configuration for KafkaChannel"
     kubectl patch configmap/kafka-channel-config \
       -n knative-eventing \
       --type merge \
-      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9095", "auth.secret.ref.name": "strimzi-sasl-plain-secret", "auth.secret.ref.namespace": "knative-eventing"}}'
+      -p '{"data":{"bootstrap.servers":"my-cluster-kafka-bootstrap.kafka:9095", "auth.secret.ref.name": "strimzi-sasl-plain-secret-legacy", "auth.secret.ref.namespace": "knative-eventing"}}'
   else
     echo "Setting up no auth configuration for KafkaChannel"
     kubectl patch configmap/kafka-channel-config \
diff --git a/test/kafka/kafka_setup.sh b/test/kafka/kafka_setup.sh
index a4ed609d0..a9789d56f 100755
--- a/test/kafka/kafka_setup.sh
+++ b/test/kafka/kafka_setup.sh
@@ -106,6 +106,13 @@ function create_sasl_secrets() {
     --from-literal=saslType="SCRAM-SHA-512" \
     --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 
+  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-secret-legacy \
+    --from-literal=ca.crt="$STRIMZI_CRT" \
+    --from-literal=password="$SASL_PASSWD" \
+    --from-literal=user="my-sasl-user" \
+    --from-literal=saslType="SCRAM-SHA-512" \
+    --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
+
   kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-plain-secret \
     --from-literal=password="$SASL_PASSWD" \
     --from-literal=user="my-sasl-user" \
@@ -113,6 +120,12 @@ function create_sasl_secrets() {
     --from-literal=sasl.mechanism="SCRAM-SHA-512" \
     --from-literal=saslType="SCRAM-SHA-512" \
     --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
+
+  kubectl create secret --namespace "${SYSTEM_NAMESPACE}" generic strimzi-sasl-plain-secret-legacy \
+    --from-literal=password="$SASL_PASSWD" \
+    --from-literal=username="my-sasl-user" \
+    --from-literal=saslType="SCRAM-SHA-512" \
+    --dry-run=client -o yaml | kubectl apply -n "${SYSTEM_NAMESPACE}" -f -
 }
 
 wait_until_pods_running kafka || fail_test "Failed to start up a Kafka cluster"