Repository: knative-sandbox/eventing-kafka-broker
Pull request #2229 (2022-05-27 14:11:17)

[release-1.4] Disable Sarama metrics since they are not exposed (#2225)
Cherry-pick of https://github.com/knative-sandbox/eventing-kafka-broker/pull/2225

Sarama metrics consume a lot of memory but we don't use/expose them.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

pierDipi: /cherry-pick release-1.3
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.3 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2229#issuecomment-1139657665):

>/cherry-pick release-1.3


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.
</details>
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2229#" title="Author self-approved">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->
pierDipi: /cherry-pick release-1.2
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.2 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2229#issuecomment-1139657756):

>/cherry-pick release-1.2


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.
</details>
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2229](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0193492) into [release-1.4](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/707941c08058fa3233abea42836d39b6cf898d1f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (707941c) will **increase** coverage by `0.01%`.
> The diff coverage is `100.00%`.

```diff
@@               Coverage Diff               @@
##           release-1.4    #2229      +/-   ##
===============================================
+ Coverage        61.27%   61.29%   +0.01%     
===============================================
  Files               70       70              
  Lines             6849     6852       +3     
===============================================
+ Hits              4197     4200       +3     
  Misses            2354     2354              
  Partials           298      298              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229?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/kafka/client.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvY2xpZW50Lmdv) | `13.04% <100.00%> (+13.04%)` | :arrow_up: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `76.52% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229?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/2229?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [707941c...0193492](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2229?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]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2229#pullrequestreview-988858893" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2229#" title="Author self-approved">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->
pierDipi: /check-cla
knative-prow[bot]: New changes are detected. LGTM label has been removed.
pierDipi: /cc @matzew 

I had to force push since the Google CLA status was stuck

Patch:
diff --git a/control-plane/pkg/kafka/client.go b/control-plane/pkg/kafka/client.go
index b129275b5..5ded6b723 100644
--- a/control-plane/pkg/kafka/client.go
+++ b/control-plane/pkg/kafka/client.go
@@ -18,8 +18,14 @@ package kafka
 
 import (
 	"github.com/Shopify/sarama"
+	"github.com/rcrowley/go-metrics"
 )
 
+func init() {
+	// Disable Sarama metrics
+	metrics.UseNilMetrics = true
+}
+
 type ConfigOption func(config *sarama.Config) error
 
 func Options(config *sarama.Config, options ...ConfigOption) error {
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 != ""
+}
diff --git a/go.mod b/go.mod
index 58e06b99e..732e0e4f0 100644
--- a/go.mod
+++ b/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/pierdipi/sacura v0.0.0-20210302185533-982357fc042b
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
 	github.com/rickb777/date v1.14.1
 	github.com/stretchr/testify v1.7.0
 	github.com/xdg-go/scram v1.1.0
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 633e2ba22..5a140aab8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -345,6 +345,7 @@ github.com/prometheus/procfs/internal/util
 github.com/prometheus/statsd_exporter/pkg/mapper
 github.com/prometheus/statsd_exporter/pkg/mapper/fsm
 # github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
+## explicit
 github.com/rcrowley/go-metrics
 # github.com/rickb777/date v1.14.1
 ## explicit


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 a50ced986..f612522f0 100644
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -283,13 +283,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"