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

Issue:
Kafka SASLType not hondored in saramaConfig in kafka distributed mode
**Describe the bug**

Kafka users that requires different SASL Mechanisms will not honor the configuration in `saramaConfig` due to it being statically set to `PLAIN` method. 
This results in users that requires `SCRAM-SHA-256` or `SCRAM-SHA-512` to be denied access to kafka.

The receiver will go in a CrashLoopBackoff due to authentication failure.

 **Expected behavior**
It should be possible to configure the SASLType (/`Mechanism`) per kafka user.

 **To Reproduce**
Requires: A kafka user that requires `SCRAM-SHA-[256|512]` SASL.
1. Configure Mechanism in `config-kafka` configmap. This will have no effect. 

```
apiVersion: v1
data:
  sarama: |
    Version: 2.0.0 # Kafka Version Compatibility From Sarama's Supported List (Major.Minor.Patch)
    Net:
      SASL:
        Enable: true
        Mechanism: SCRAM-SHA-512
        Version: 1
```
2. `kafka-cluster-*-receiver-*` pod will CrashLoop with the following error message:
```
[sarama] 2021/01/23 10:18:37 Initializing new client
[sarama] 2021/01/23 10:18:37 client/metadata fetching metadata for all topics from broker hostname:9092
[sarama] 2021/01/23 10:18:37 Successful SASL handshake. Available mechanisms: [SCRAM-SHA-512 PLAIN]
[sarama] 2021/01/23 10:18:38 Error returned from broker during SASL flow hostname9092: kafka server: SASL Authentication failed.
[sarama] 2021/01/23 10:18:38 Closed connection to broker hostname:9092
[sarama] 2021/01/23 10:18:38 client/metadata failed SASL authentication
[sarama] 2021/01/23 10:18:38 Closing Client
```

 **Knative release version**
nightly

 **Additional context**


Pull request description:
#331 Add sasltype as optional secret parameter for kafka user configuration
Fixes #331

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add `sasltype` to kafka secret to configure optional sasltype as `SCRAM-SHA-256` or `SCRAM-SHA-512`.

**Release Note**

<!--
🗒️ 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
Adding new optional field named `sasltype` to default kafka-secret to enable other Kafka SASL Methods than `PLAIN`. Supports `SCRAM-SHA-256` or `SCRAM-SHA-512`.
```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 145a97456..5b8f750bb 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -79,6 +79,7 @@ func main() {
 			SASL: &client.KafkaSaslConfig{
 				User:     environment.KafkaUsername,
 				Password: environment.KafkaPassword,
+				SaslType: environment.KafkaSaslType,
 			},
 		}
 	}
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index 1243141cf..a54ae3840 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -82,6 +82,7 @@ func main() {
 			SASL: &client.KafkaSaslConfig{
 				User:     environment.KafkaUsername,
 				Password: environment.KafkaPassword,
+				SaslType: environment.KafkaSaslType,
 			},
 		}
 	}
diff --git a/config/channel/distributed/300-eventing-kafka-configmap.yaml b/config/channel/distributed/300-eventing-kafka-configmap.yaml
index 59a726455..76e507e6b 100644
--- a/config/channel/distributed/300-eventing-kafka-configmap.yaml
+++ b/config/channel/distributed/300-eventing-kafka-configmap.yaml
@@ -25,7 +25,6 @@ data:
         Enable: true
       SASL:
         Enable: true
-        Mechanism: PLAIN
         Version: 1
     Metadata:
       RefreshFrequency: 300000000000  # 5 minutes
diff --git a/config/channel/distributed/300-kafka-secret.yaml b/config/channel/distributed/300-kafka-secret.yaml
index 70744b6d2..01a527e95 100644
--- a/config/channel/distributed/300-kafka-secret.yaml
+++ b/config/channel/distributed/300-kafka-secret.yaml
@@ -19,6 +19,7 @@ data:
   namespace: ""
   password: ""
   username: ""
+  sasltype: ""
 kind: Secret
 metadata:
   name: kafka-cluster
diff --git a/config/channel/distributed/README.md b/config/channel/distributed/README.md
index 36a868737..170b8c914 100644
--- a/config/channel/distributed/README.md
+++ b/config/channel/distributed/README.md
@@ -126,6 +126,10 @@ your Kafka cluster.
   - **Net.SASL.Password:** If you specify the password in the ConfigMap it will
     be overridden by the values from the
     [kafka-secret.yaml](300-kafka-secret.yaml) file!
+  - **Net.SASL.Mechanism:** If you specify the Mechanism in the ConfigMap it
+    will be overridden by the value of `sasltype` from the
+    [kafka-secret.yaml](300-kafka-secret.yaml) or default to `PLAIN`.
+    Optional values are `SCRAM-SHA-256` and `SCRAM-SHA-512`.
   - **Net.TLS.Enable** Enable (true) / disable (false) according to your
     authentication needs.
   - **Net.TLS.Config:** The Golang
diff --git a/config/source/200-serviceaccount.yaml b/config/source/single/200-serviceaccount.yaml
similarity index 100%
rename from config/source/200-serviceaccount.yaml
rename to config/source/single/200-serviceaccount.yaml
diff --git a/config/source/201-clusterrole.yaml b/config/source/single/201-clusterrole.yaml
similarity index 100%
rename from config/source/201-clusterrole.yaml
rename to config/source/single/201-clusterrole.yaml
diff --git a/config/source/202-clusterrolebinding.yaml b/config/source/single/202-clusterrolebinding.yaml
similarity index 100%
rename from config/source/202-clusterrolebinding.yaml
rename to config/source/single/202-clusterrolebinding.yaml
diff --git a/config/source/300-kafkabinding.yaml b/config/source/single/300-kafkabinding.yaml
similarity index 100%
rename from config/source/300-kafkabinding.yaml
rename to config/source/single/300-kafkabinding.yaml
diff --git a/config/source/300-kafkasource.yaml b/config/source/single/300-kafkasource.yaml
similarity index 100%
rename from config/source/300-kafkasource.yaml
rename to config/source/single/300-kafkasource.yaml
diff --git a/config/source/400-controller-service.yaml b/config/source/single/400-controller-service.yaml
similarity index 100%
rename from config/source/400-controller-service.yaml
rename to config/source/single/400-controller-service.yaml
diff --git a/config/source/500-controller.yaml b/config/source/single/500-controller.yaml
similarity index 100%
rename from config/source/500-controller.yaml
rename to config/source/single/500-controller.yaml
diff --git a/config/source/500-webhook-configuration.yaml b/config/source/single/500-webhook-configuration.yaml
similarity index 100%
rename from config/source/500-webhook-configuration.yaml
rename to config/source/single/500-webhook-configuration.yaml
diff --git a/config/source/README.md b/config/source/single/README.md
similarity index 100%
rename from config/source/README.md
rename to config/source/single/README.md
diff --git a/config/source/deployments/controller-service.yaml b/config/source/single/deployments/controller-service.yaml
similarity index 100%
rename from config/source/deployments/controller-service.yaml
rename to config/source/single/deployments/controller-service.yaml
diff --git a/config/source/deployments/controller.yaml b/config/source/single/deployments/controller.yaml
similarity index 100%
rename from config/source/deployments/controller.yaml
rename to config/source/single/deployments/controller.yaml
diff --git a/config/source/resources/kafkabinding.yaml b/config/source/single/resources/kafkabinding.yaml
similarity index 100%
rename from config/source/resources/kafkabinding.yaml
rename to config/source/single/resources/kafkabinding.yaml
diff --git a/config/source/resources/kafkasource.yaml b/config/source/single/resources/kafkasource.yaml
similarity index 100%
rename from config/source/resources/kafkasource.yaml
rename to config/source/single/resources/kafkasource.yaml
diff --git a/config/source/roles/clusterrole.yaml b/config/source/single/roles/clusterrole.yaml
similarity index 100%
rename from config/source/roles/clusterrole.yaml
rename to config/source/single/roles/clusterrole.yaml
diff --git a/config/source/webhooks/kafka.yaml b/config/source/single/webhooks/kafka.yaml
similarity index 100%
rename from config/source/webhooks/kafka.yaml
rename to config/source/single/webhooks/kafka.yaml
diff --git a/hack/release.sh b/hack/release.sh
index 4ab5b20b8..0f264fd76 100755
--- a/hack/release.sh
+++ b/hack/release.sh
@@ -26,7 +26,7 @@ declare -A COMPONENTS
 COMPONENTS=(
   ["channel-consolidated.yaml"]="config/channel/consolidated"
   ["channel-distributed.yaml"]="config/channel/distributed"
-  ["source.yaml"]="config/source"
+  ["source.yaml"]="config/source/single"
 )
 readonly COMPONENTS
 
diff --git a/pkg/channel/distributed/common/env/constants.go b/pkg/channel/distributed/common/env/constants.go
index 575d9baf3..58a615ff8 100644
--- a/pkg/channel/distributed/common/env/constants.go
+++ b/pkg/channel/distributed/common/env/constants.go
@@ -32,6 +32,7 @@ const (
 	KafkaBrokerEnvVarKey   = "KAFKA_BROKERS"
 	KafkaUsernameEnvVarKey = "KAFKA_USERNAME"
 	KafkaPasswordEnvVarKey = "KAFKA_PASSWORD"
+	KafkaSaslTypeEnvVarKey = "KAKFA_SASL_TYPE"
 
 	// Kafka Configuration
 	KafkaTopicEnvVarKey = "KAFKA_TOPIC"
diff --git a/pkg/channel/distributed/common/kafka/constants/constants.go b/pkg/channel/distributed/common/kafka/constants/constants.go
index f9a94aa0a..15a5d6589 100644
--- a/pkg/channel/distributed/common/kafka/constants/constants.go
+++ b/pkg/channel/distributed/common/kafka/constants/constants.go
@@ -33,6 +33,7 @@ const (
 	KafkaSecretKeyNamespace = "namespace"
 	KafkaSecretKeyUsername  = "username"
 	KafkaSecretKeyPassword  = "password"
+	KafkaSecretKeySaslType  = "sasltype"
 
 	// Kafka Admin/Consumer/Producer Config Values
 	ConfigNetSaslVersion = sarama.SASLHandshakeV1 // Latest version, seems to work with EventHubs as well.
diff --git a/pkg/channel/distributed/controller/constants/constants.go b/pkg/channel/distributed/controller/constants/constants.go
index 6f17cff70..dc7297f0a 100644
--- a/pkg/channel/distributed/controller/constants/constants.go
+++ b/pkg/channel/distributed/controller/constants/constants.go
@@ -51,6 +51,7 @@ const (
 	KafkaSecretDataKeyBrokers  = "brokers"
 	KafkaSecretDataKeyUsername = "username"
 	KafkaSecretDataKeyPassword = "password"
+	KafkaSecretDataKeySaslType = "sasltype"
 
 	// Prometheus MetricsPort
 	MetricsPortName = "metrics"
diff --git a/pkg/channel/distributed/controller/kafkachannel/controller.go b/pkg/channel/distributed/controller/kafkachannel/controller.go
index 1acb661ac..8d71ec04f 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller.go
@@ -83,6 +83,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	kafkaBrokers := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyBrokers])
 	kafkaUsername := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyUsername])
 	kafkaPassword := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyPassword])
+	kafkaSaslType := string(kafkaSecret.Data[clientconstants.KafkaSecretKeySaslType])
 
 	// Create A Kafka Auth Config From Current Credentials (Secret Data Takes Precedence Over ConfigMap)
 	var kafkaAuthCfg *commonclient.KafkaAuthConfig
@@ -91,6 +92,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 			SASL: &commonclient.KafkaSaslConfig{
 				User:     kafkaUsername,
 				Password: kafkaPassword,
+				SaslType: kafkaSaslType,
 			},
 		}
 	}
@@ -137,6 +139,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 		kafkaBrokers:         kafkaBrokers,
 		kafkaUsername:        kafkaUsername,
 		kafkaPassword:        kafkaPassword,
+		kafkaSaslType:        kafkaSaslType,
 	}
 
 	// Watch The Settings ConfigMap For Changes
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index 37ed3590c..bfc752201 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -581,6 +581,19 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 				},
 			},
 		})
+
+		// Append The Kafka SaslType As Env Var
+		optional := true
+		envVars = append(envVars, corev1.EnvVar{
+			Name: commonenv.KafkaSaslTypeEnvVarKey,
+			ValueFrom: &corev1.EnvVarSource{
+				SecretKeyRef: &corev1.SecretKeySelector{
+					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
+					Key:                  constants.KafkaSecretDataKeySaslType,
+					Optional:             &optional,
+				},
+			},
+		})
 	}
 
 	// Return The Dispatcher Deployment EnvVars Array
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler.go b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
index b445e90c8..779e53cf4 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler.go
@@ -67,6 +67,7 @@ type Reconciler struct {
 	kafkaBrokers         string
 	kafkaUsername        string
 	kafkaPassword        string
+	kafkaSaslType        string
 }
 
 var (
@@ -275,6 +276,7 @@ func (r *Reconciler) configMapObserver(ctx context.Context, configMap *corev1.Co
 			SASL: &client.KafkaSaslConfig{
 				User:     r.kafkaUsername,
 				Password: r.kafkaPassword,
+				SaslType: r.kafkaSaslType,
 			},
 		}
 	}
diff --git a/pkg/channel/distributed/controller/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index 421bfa653..ebb581241 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -442,6 +442,18 @@ func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.
 		},
 	})
 
+	optional := true
+	envVars = append(envVars, corev1.EnvVar{
+		Name: commonenv.KafkaSaslTypeEnvVarKey,
+		ValueFrom: &corev1.EnvVarSource{
+			SecretKeyRef: &corev1.SecretKeySelector{
+				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
+				Key:                  constants.KafkaSecretDataKeySaslType,
+				Optional:             &optional,
+			},
+		},
+	})
+
 	// Return The Receiver Deployment EnvVars Array
 	return envVars, nil
 }
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 186975258..f0be4365a 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -297,6 +297,7 @@ func (d *DispatcherImpl) ConfigChanged(ctx context.Context, configMap *corev1.Co
 				SASL: &client.KafkaSaslConfig{
 					User:     d.SaramaConfig.Net.SASL.User,
 					Password: d.SaramaConfig.Net.SASL.Password,
+					SaslType: string(d.SaramaConfig.Net.SASL.Mechanism),
 				},
 			}
 			configBuilder = configBuilder.WithAuth(kafkaAuthCfg)
diff --git a/pkg/channel/distributed/dispatcher/env/environment.go b/pkg/channel/distributed/dispatcher/env/environment.go
index 27f1d5562..4071afabb 100644
--- a/pkg/channel/distributed/dispatcher/env/environment.go
+++ b/pkg/channel/distributed/dispatcher/env/environment.go
@@ -54,6 +54,7 @@ type Environment struct {
 	// Kafka Authorization
 	KafkaUsername string // Optional
 	KafkaPassword string // Optional
+	KafkaSaslType string // Optional
 }
 
 // Get The Environment
@@ -142,6 +143,9 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// Get The Optional KafkaPassword Config Value
 	environment.KafkaPassword = env.GetOptionalConfigValue(logger, env.KafkaPasswordEnvVarKey, "")
 
+	// Get The Optional KafkaSaslType Config Value
+	environment.KafkaSaslType = env.GetOptionalConfigValue(logger, env.KafkaSaslTypeEnvVarKey, "")
+
 	// Clone The Environment & Mask The Password For Safe Logging
 	safeEnvironment := *environment
 	if len(safeEnvironment.KafkaPassword) > 0 {
diff --git a/pkg/channel/distributed/receiver/env/environment.go b/pkg/channel/distributed/receiver/env/environment.go
index 73f87eb08..a9f0a9975 100644
--- a/pkg/channel/distributed/receiver/env/environment.go
+++ b/pkg/channel/distributed/receiver/env/environment.go
@@ -52,6 +52,7 @@ type Environment struct {
 	// Kafka Authorization
 	KafkaUsername string // Optional
 	KafkaPassword string // Optional
+	KafkaSaslType string // Optional
 }
 
 // Get The Environment
@@ -128,6 +129,9 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	// Get The Optional KafkaPassword Config Value
 	environment.KafkaPassword = env.GetOptionalConfigValue(logger, env.KafkaPasswordEnvVarKey, "")
 
+	// Get the Optional KafkaSaslType Config value
+	environment.KafkaSaslType = env.GetOptionalConfigValue(logger, env.KafkaSaslTypeEnvVarKey, "")
+
 	// Clone The Environment & Mask The Password For Safe Logging
 	safeEnvironment := *environment
 	if len(safeEnvironment.KafkaPassword) > 0 {
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 4a4ec5406..b18899785 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -198,6 +198,7 @@ func (p *Producer) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMa
 				SASL: &client.KafkaSaslConfig{
 					User:     p.configuration.Net.SASL.User,
 					Password: p.configuration.Net.SASL.Password,
+					SaslType: string(p.configuration.Net.SASL.Mechanism),
 				},
 			}
 			configBuilder = configBuilder.WithAuth(kafkaAuthCfg)


Test patch:
diff --git a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
index 7b622cd10..12dc88a4c 100644
--- a/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
+++ b/pkg/channel/distributed/controller/kafkachannel/reconciler_test.go
@@ -676,6 +676,7 @@ func TestReconcile(t *testing.T) {
 			kafkaSecret:          controllertesting.KafkaSecretName,
 			kafkaUsername:        controllertesting.KafkaSecretDataValueUsername,
 			kafkaPassword:        controllertesting.KafkaSecretDataValuePassword,
+			kafkaSaslType:        controllertesting.KafkaSecretDataValueSaslType,
 		}
 		return kafkachannelreconciler.NewReconciler(ctx, logger, r.kafkaClientSet, listers.GetKafkaChannelLister(), controller.GetEventRecorder(ctx), r)
 	}, logger.Desugar()))
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index d3bf8e50b..42f1eb86e 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -80,6 +80,7 @@ const (
 	KafkaSecretDataValueBrokers  = "TestKafkaSecretDataBrokers"
 	KafkaSecretDataValueUsername = "TestKafkaSecretDataUsername"
 	KafkaSecretDataValuePassword = "TestKafkaSecretDataPassword"
+	KafkaSecretDataValueSaslType = "PLAIN"
 
 	// ChannelSpec Test Data
 	NumPartitions     = 123
@@ -289,6 +290,7 @@ func NewKafkaSecret(options ...KafkaSecretOption) *corev1.Secret {
 			constants.KafkaSecretDataKeyBrokers:  []byte(KafkaSecretDataValueBrokers),
 			constants.KafkaSecretDataKeyUsername: []byte(KafkaSecretDataValueUsername),
 			constants.KafkaSecretDataKeyPassword: []byte(KafkaSecretDataValuePassword),
+			constants.KafkaSecretDataKeySaslType: []byte(KafkaSecretDataValueSaslType),
 		},
 		Type: "opaque",
 	}
@@ -577,6 +579,9 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 	// Replicas Int Reference
 	replicas := int32(ReceiverReplicas)
 
+	// Optional EnvVar
+	optional := true
+
 	// Create The Receiver Deployment
 	deployment := &appsv1.Deployment{
 		TypeMeta: metav1.TypeMeta{
@@ -707,6 +712,16 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 										},
 									},
 								},
+								{
+									Name: commonenv.KafkaSaslTypeEnvVarKey,
+									ValueFrom: &corev1.EnvVarSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
+											Key:                  constants.KafkaSecretDataKeySaslType,
+											Optional:             &optional,
+										},
+									},
+								},
 							},
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
@@ -791,6 +806,7 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 	dispatcherName := util.DispatcherDnsSafeName(sparseKafkaChannel)
 	topicName := util.TopicName(sparseKafkaChannel)
 	systemNamespace := system.Namespace()
+	optional := true
 
 	// Replicas Int Reference
 	replicas := int32(DispatcherReplicas)
@@ -926,6 +942,16 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 										},
 									},
 								},
+								{
+									Name: commonenv.KafkaSaslTypeEnvVarKey,
+									ValueFrom: &corev1.EnvVarSource{
+										SecretKeyRef: &corev1.SecretKeySelector{
+											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
+											Key:                  constants.KafkaSecretDataKeySaslType,
+											Optional:             &optional,
+										},
+									},
+								},
 							},
 							ImagePullPolicy: corev1.PullIfNotPresent,
 							Resources: corev1.ResourceRequirements{
diff --git a/pkg/channel/distributed/dispatcher/env/environment_test.go b/pkg/channel/distributed/dispatcher/env/environment_test.go
index bfdcf9d42..eead5cdef 100644
--- a/pkg/channel/distributed/dispatcher/env/environment_test.go
+++ b/pkg/channel/distributed/dispatcher/env/environment_test.go
@@ -44,6 +44,7 @@ const (
 	serviceName     = "TestServiceName"
 	kafkaUsername   = "TestKafkaUsername"
 	kafkaPassword   = "TestKafkaPassword"
+	kafkaSaslType   = "SCRAM-SHA-512"
 	podName         = "TestPod"
 	containerName   = "TestContainer"
 )
@@ -62,6 +63,7 @@ type TestCase struct {
 	serviceName          string
 	kafkaUsername        string
 	kafkaPassword        string
+	kafkaSaslType        string
 	podName              string
 	containerName        string
 	expectedError        error
@@ -166,6 +168,7 @@ func TestGetEnvironment(t *testing.T) {
 			assertSetenv(t, commonenv.ServiceNameEnvVarKey, testCase.serviceName)
 			assertSetenv(t, commonenv.KafkaUsernameEnvVarKey, testCase.kafkaUsername)
 			assertSetenv(t, commonenv.KafkaPasswordEnvVarKey, testCase.kafkaPassword)
+			assertSetenv(t, commonenv.KafkaSaslTypeEnvVarKey, testCase.kafkaSaslType)
 			assertSetenv(t, commonenv.PodNameEnvVarKey, testCase.podName)
 			assertSetenv(t, commonenv.ContainerNameEnvVarKey, testCase.containerName)
 
@@ -186,6 +189,7 @@ func TestGetEnvironment(t *testing.T) {
 				assert.Equal(t, testCase.serviceName, environment.ServiceName)
 				assert.Equal(t, testCase.kafkaUsername, environment.KafkaUsername)
 				assert.Equal(t, testCase.kafkaPassword, environment.KafkaPassword)
+				assert.Equal(t, testCase.kafkaSaslType, environment.KafkaSaslType)
 				assert.Equal(t, testCase.podName, environment.PodName)
 				assert.Equal(t, testCase.containerName, environment.ContainerName)
 				assert.Equal(t, testCase.expectedResyncPeriod, strconv.Itoa(int(environment.ResyncPeriod/time.Minute)))
@@ -223,6 +227,7 @@ func getValidTestCase(name string) TestCase {
 		serviceName:          serviceName,
 		kafkaUsername:        kafkaUsername,
 		kafkaPassword:        kafkaPassword,
+		kafkaSaslType:        kafkaSaslType,
 		podName:              podName,
 		containerName:        containerName,
 		expectedError:        nil,
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index f711c8152..0fd29c8f2 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -104,7 +104,7 @@ readonly KAFKA_SOURCE_CRD_CONFIG_DIR="$(mktemp -d)"
 trap "{ for dirrm in \"${KAFKA_CRD_CONFIG_DIR}\" \"${KAFKA_SOURCE_CRD_CONFIG_DIR}\"; do rm \"\${dirrm}\"/*; rmdir \"\${dirrm}\"; done }" EXIT
 
 # Kafka channel CRD config template directory.
-readonly KAFKA_SOURCE_TEMPLATE_DIR="config/source"
+readonly KAFKA_SOURCE_TEMPLATE_DIR="config/source/single"
 
 # Namespaces where we install Eventing components
 # This is the namespace of knative-eventing itself