Repository: knative-sandbox/eventing-kafka
Pull request #395 (2021-02-16 13:07:24)

Adding HPA for webhook
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Similar to "eventing core" (see https://github.com/knative/eventing/pull/4792/files), I am adding HPA for the webhook.

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**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 HorizontalPodAutoscaler and PodDisruptionBudget for the kafka webhook

```

**Docs**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/395#" title="Author self-approved">matzew</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).

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/blob/master/OWNERS)~~ [matzew]

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":[]} -->
matzew: /hold

/assign @travis-minke-sap 
/assign @antoineco 

Starting w/ the simple webhhook for HPA
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=h1) Report
> Merging [#395](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=desc) (14b4ab1) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/587a2081b8bae4dfd93c523efb3421b4f361a5cc?el=desc) (587a208) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           master     #395   +/-   ##
=======================================
  Coverage   73.09%   73.09%           
=======================================
  Files         128      128           
  Lines        4973     4973           
=======================================
  Hits         3635     3635           
  Misses       1106     1106           
  Partials      232      232           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=footer). Last update [587a208...f6218c0](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/395?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/395#pullrequestreview-591257657" title="LGTM">antoineco</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/395#" title="Author self-approved">matzew</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).

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/blob/master/OWNERS)~~ [matzew]

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":[]} -->
matzew: Leaving the hold for @travis-minke-sap 

Patch:
diff --git a/.github/workflows/knative-releasability.yaml b/.github/workflows/knative-releasability.yaml
index 3ff9d2df7..f9b1799f2 100644
--- a/.github/workflows/knative-releasability.yaml
+++ b/.github/workflows/knative-releasability.yaml
@@ -157,3 +157,11 @@ jobs:
             ${{ env.CHECK_MESSAGE }}
             ${{ env.VERIFY_MESSAGE }}
             For detailed logs: https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}
+
+      - name: Fail if NO-GO
+        if: steps.exists.outputs.release-branch == 'false' && env.current == 'false'
+        run: |
+          # When we have figured out that things are NO-GO, we intentionally fail the job
+          # so that the status badge shows up red and we can use the badges to create a
+          # releasability dashboard for all of the repos.
+          exit 1
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 4fa74bd2c..568a992e5 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -40,7 +40,6 @@ import (
 	dispatcherhealth "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/health"
 	"knative.dev/eventing-kafka/pkg/client/clientset/versioned"
 	"knative.dev/eventing-kafka/pkg/client/informers/externalversions"
-	"knative.dev/eventing-kafka/pkg/common/client"
 	kncontroller "knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	eventingmetrics "knative.dev/pkg/metrics"
@@ -74,16 +73,10 @@ func main() {
 		logger.Fatal("Failed To Load Environment Variables - Terminating!", zap.Error(err))
 	}
 
-	// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-	var kafkaAuthCfg *client.KafkaAuthConfig
-	if environment.KafkaUsername != "" {
-		kafkaAuthCfg = &client.KafkaAuthConfig{
-			SASL: &client.KafkaSaslConfig{
-				User:     environment.KafkaUsername,
-				Password: environment.KafkaPassword,
-				SaslType: environment.KafkaSaslType,
-			},
-		}
+	// Update The Sarama Config - Username/Password Overrides (Values From Secret Take Precedence Over ConfigMap)
+	kafkaAuthCfg, err := commonconfig.GetAuthConfigFromKubernetes(ctx, environment.KafkaSecretName, environment.KafkaSecretNamespace)
+	if err != nil {
+		logger.Fatal("Failed To Load Auth Config", zap.Error(err))
 	}
 
 	// Load The Sarama & Eventing-Kafka Configuration From The ConfigMap
@@ -126,10 +119,10 @@ func main() {
 	dispatcherConfig := dispatch.DispatcherConfig{
 		Logger:        logger,
 		ClientId:      constants.Component,
-		Brokers:       strings.Split(environment.KafkaBrokers, ","),
+		Brokers:       strings.Split(ekConfig.Kafka.Brokers, ","),
 		Topic:         environment.KafkaTopic,
-		Username:      environment.KafkaUsername,
-		Password:      environment.KafkaPassword,
+		Username:      kafkaAuthCfg.SASL.User,
+		Password:      kafkaAuthCfg.SASL.Password,
 		ChannelKey:    environment.ChannelKey,
 		StatsReporter: statsReporter,
 		SaramaConfig:  saramaConfig,
@@ -142,6 +135,12 @@ func main() {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
 
+	// Watch The Secret For Changes
+	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	if err != nil {
+		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
+	}
+
 	config, err := clientcmd.BuildConfigFromFlags(*serverURL, *kubeconfig)
 	if err != nil {
 		logger.Fatal("Error building kubeconfig", zap.Error(err))
@@ -221,6 +220,31 @@ func configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
 	newDispatcher := dispatcher.ConfigChanged(ctx, configMap)
 	if newDispatcher != nil {
 		// The configuration change caused a new dispatcher to be created, so switch to that one
+		logger.Info("Config Changed; Dispatcher Reconfigured")
+		dispatcher = newDispatcher
+	}
+}
+
+// secretObserver is the callback function that handles changes to our Secret
+func secretObserver(ctx context.Context, secret *corev1.Secret) {
+	logger := logging.FromContext(ctx)
+
+	if secret == nil {
+		logger.Warn("Nil Secret passed to secretObserver; ignoring")
+		return
+	}
+
+	if dispatcher == nil {
+		// This typically happens during startup
+		logger.Debug("Dispatcher is nil during call to secretObserver; ignoring changes")
+		return
+	}
+
+	// Toss the new secret to the dispatcher for inspection and action
+	newDispatcher := dispatcher.SecretChanged(ctx, secret)
+	if newDispatcher != nil {
+		// The configuration change caused a new dispatcher to be created, so switch to that one
+		logger.Info("Secret Changed; Dispatcher Reconfigured")
 		dispatcher = newDispatcher
 	}
 }
diff --git a/cmd/channel/distributed/receiver/main.go b/cmd/channel/distributed/receiver/main.go
index a54ae3840..65add32e2 100644
--- a/cmd/channel/distributed/receiver/main.go
+++ b/cmd/channel/distributed/receiver/main.go
@@ -37,7 +37,6 @@ import (
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/env"
 	channelhealth "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/health"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/receiver/producer"
-	"knative.dev/eventing-kafka/pkg/common/client"
 	eventingchannel "knative.dev/eventing/pkg/channel"
 	"knative.dev/pkg/kmeta"
 	"knative.dev/pkg/logging"
@@ -74,17 +73,10 @@ func main() {
 		logger.Fatal("Invalid / Missing Environment Variables - Terminating", zap.Error(err))
 	}
 
-	var kafkaAuthCfg *client.KafkaAuthConfig
-
-	if environment.KafkaUsername != "" {
-		// Update The Sarama Config - Username/Password Overrides (EnvVars From Secret Take Precedence Over ConfigMap)
-		kafkaAuthCfg = &client.KafkaAuthConfig{
-			SASL: &client.KafkaSaslConfig{
-				User:     environment.KafkaUsername,
-				Password: environment.KafkaPassword,
-				SaslType: environment.KafkaSaslType,
-			},
-		}
+	// Update The Sarama Config - Username/Password Overrides (Values From Secret Take Precedence Over ConfigMap)
+	kafkaAuthCfg, err := commonconfig.GetAuthConfigFromKubernetes(ctx, environment.KafkaSecretName, environment.KafkaSecretNamespace)
+	if err != nil {
+		logger.Fatal("Failed To Load Auth Config", zap.Error(err))
 	}
 
 	// Load The Sarama & Eventing-Kafka Configuration From The ConfigMap
@@ -130,8 +122,14 @@ func main() {
 		logger.Fatal("Failed To Initialize ConfigMap Watcher", zap.Error(err))
 	}
 
+	// Watch The Secret For Changes
+	err = commonconfig.InitializeSecretWatcher(ctx, environment.KafkaSecretNamespace, environment.KafkaSecretName, secretObserver)
+	if err != nil {
+		logger.Fatal("Failed To Start Secret Watcher", zap.Error(err))
+	}
+
 	// Initialize The Kafka Producer In Order To Start Processing Status Events
-	kafkaProducer, err = producer.NewProducer(logger, saramaConfig, strings.Split(environment.KafkaBrokers, ","), statsReporter, healthServer)
+	kafkaProducer, err = producer.NewProducer(logger, saramaConfig, strings.Split(ekConfig.Kafka.Brokers, ","), statsReporter, healthServer)
 	if err != nil {
 		logger.Fatal("Failed To Initialize Kafka Producer", zap.Error(err))
 	}
@@ -206,6 +204,7 @@ func configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
 		logger.Warn("Nil ConfigMap passed to configMapObserver; ignoring")
 		return
 	}
+
 	if kafkaProducer == nil {
 		// This typically happens during startup
 		logger.Debug("Producer is nil during call to configMapObserver; ignoring changes")
@@ -216,7 +215,31 @@ func configMapObserver(ctx context.Context, configMap *corev1.ConfigMap) {
 	newProducer := kafkaProducer.ConfigChanged(ctx, configMap)
 	if newProducer != nil {
 		// The configuration change caused a new producer to be created, so switch to that one
-		logger.Info("Producer Reconfigured; Switching To New Producer")
+		logger.Info("Config Changed; Receiver Reconfigured")
+		kafkaProducer = newProducer
+	}
+}
+
+// secretObserver is the callback function that handles changes to our Secret
+func secretObserver(ctx context.Context, secret *corev1.Secret) {
+	logger := logging.FromContext(ctx)
+
+	if secret == nil {
+		logger.Warn("Nil Secret passed to secretObserver; ignoring")
+		return
+	}
+
+	if kafkaProducer == nil {
+		// This typically happens during startup
+		logger.Debug("Producer is nil during call to secretObserver; ignoring changes")
+		return
+	}
+
+	// Toss the new secret to the producer for inspection and action
+	newProducer := kafkaProducer.SecretChanged(ctx, secret)
+	if newProducer != nil {
+		// The configuration change caused a new producer to be created, so switch to that one
+		logger.Info("Secret Changed; Receiver Reconfigured")
 		kafkaProducer = newProducer
 	}
 }
diff --git a/config/channel/consolidated/500-webhook-hpa.yaml b/config/channel/consolidated/500-webhook-hpa.yaml
new file mode 120000
index 000000000..a37162493
--- /dev/null
+++ b/config/channel/consolidated/500-webhook-hpa.yaml
@@ -0,0 +1,1 @@
+../webhook/webhook-hpa.yaml
\ No newline at end of file
diff --git a/config/channel/distributed/300-eventing-kafka-configmap.yaml b/config/channel/distributed/300-eventing-kafka-configmap.yaml
index 76e507e6b..7b445a09a 100644
--- a/config/channel/distributed/300-eventing-kafka-configmap.yaml
+++ b/config/channel/distributed/300-eventing-kafka-configmap.yaml
@@ -46,6 +46,7 @@ data:
       memoryRequest: 50Mi
       replicas: 1
     kafka:
+      brokers: REPLACE_WITH_CLUSTER_URL
       enableSaramaLogging: false
       topic:
         defaultNumPartitions: 4
diff --git a/config/channel/distributed/300-kafka-secret.yaml b/config/channel/distributed/300-kafka-secret.yaml
index 01a527e95..7b29f1acb 100644
--- a/config/channel/distributed/300-kafka-secret.yaml
+++ b/config/channel/distributed/300-kafka-secret.yaml
@@ -14,8 +14,7 @@
 
 apiVersion: v1
 data:
-  # Specify Kafka Broker & Optional Credentials Here
-  brokers: ""
+  # Specify Optional Kafka Credentials Here
   namespace: ""
   password: ""
   username: ""
diff --git a/config/channel/distributed/500-webhook-hpa.yaml b/config/channel/distributed/500-webhook-hpa.yaml
new file mode 120000
index 000000000..a37162493
--- /dev/null
+++ b/config/channel/distributed/500-webhook-hpa.yaml
@@ -0,0 +1,1 @@
+../webhook/webhook-hpa.yaml
\ No newline at end of file
diff --git a/config/channel/distributed/README.md b/config/channel/distributed/README.md
index 8c7a3b1ae..380700b68 100644
--- a/config/channel/distributed/README.md
+++ b/config/channel/distributed/README.md
@@ -55,9 +55,32 @@ one of `kafka`, `azure`, or `custom` as follows...
 
 ### Install & Label Kafka Credentials In Knative-Eventing Namespace
 
-The Kafka brokers and associated auth are specified in a Kubernetes Secret in
-the `knative-eventing` namespace which has been labelled as
-`eventing-kafka.knative.dev/kafka-secret="true"`. For the `kafka` and `custom`
+The Kafka brokers are specified in a configmap provided in the config
+directory, [eventing-kafka-configmap.yaml](300-eventing-kafka-configmap.yaml), under
+the kafka.brokers entry of the eventing-kafka field.  The text "REPLACE_WITH_CLUSTER_URL"
+should be replaced with a real broker string.  An example of such a string might be:
+```
+  eventing-kafka: |
+    kafka:
+      brokers: SASL_SSL://my-cluster.eu-west-1.aws.confluent.cloud:9092
+```
+Example value for Azure Event Hubs:
+```
+  eventing-kafka: |
+    kafka:
+      brokers: my-cluster-name-1.servicebus.windows.net:9093
+```
+Example value for a multi-broker Kafka (must be base64 encoded):
+
+```
+  eventing-kafka: |
+    kafka:
+      brokers: 10.254.11.41:9093,10.254.11.42:9093,10.254.11.43:9093
+```
+
+The associated auth information is specified in a Kubernetes Secret in the
+`knative-eventing` namespace which has been labelled as
+`eventing-kafka.knative.dev/kafka-secret="true"`.  For the `kafka` and `custom`
 Admin Types (see above) there should be exactly 1 such Secret. For the `azure`
 Admin Type (see above) multiple such Secrets are possible, each representing a
 different EventHub Namespace. In that case Topics will be load balanced across
@@ -71,7 +94,6 @@ the required authentication with the Kafka cluster.
 Example values for a standard Kafka (must be base64 encoded):
 
 ```
-  brokers: SASL_SSL://my-cluster.eu-west-1.aws.confluent.cloud:9092
   password: XVLEXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
   username: KIELSXXXXXXXXXXX
 ```
@@ -80,7 +102,6 @@ Example values for Azure Event Hubs (must be base64 encoded):
 
 ```
   namespace: my-cluster-name-1
-  brokers: my-cluster-name-1.servicebus.windows.net:9093
   password: Endpoint=sb://my-cluster-name-1.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX=
   username: $ConnectionString
 ```
@@ -91,7 +112,6 @@ installation, they may also be created manually:
 ```
 # Example A Creating A Kafka Secret In Knative-Eventing
 kubectl create secret -n knative-eventing generic kafka-credentials \
-    --from-literal=brokers=<BROKER CONNECTION STRING> \
     --from-literal=username=<USERNAME> \
     --from-literal=password=<PASSWORD> \
     --from-literal=namespace=<AZURE EVENTHUBS NAMESPACE> \
@@ -175,3 +195,5 @@ your Kafka cluster.
   - **kafka.adminType:** As described above this value must be set to one of
     `kafka`, `azure`, or `custom`. The default is `kakfa` and will be used by
     most users.
+  - **kafka.brokers:** This field must be set to your kafka brokers string (see above)
+
diff --git a/config/channel/webhook/webhook-hpa.yaml b/config/channel/webhook/webhook-hpa.yaml
new file mode 100644
index 000000000..2fcb0861b
--- /dev/null
+++ b/config/channel/webhook/webhook-hpa.yaml
@@ -0,0 +1,49 @@
+# 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
+#
+#     https://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.
+
+apiVersion: autoscaling/v2beta2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: kafka-webhook
+  namespace: knative-eventing
+  labels:
+    eventing.knative.dev/release: devel
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: kafka-webhook
+  minReplicas: 1
+  maxReplicas: 5
+  metrics:
+  - type: Resource
+    resource:
+      name: cpu
+      target:
+        type: Utilization
+        averageUtilization: 100
+---
+# Webhook PDB.
+apiVersion: policy/v1beta1
+kind: PodDisruptionBudget
+metadata:
+  name: kafka-webhook
+  namespace: knative-eventing
+  labels:
+    eventing.knative.dev/release: devel
+spec:
+  minAvailable: 80%
+  selector:
+    matchLabels:
+      app: kafka-webhook
diff --git a/go.mod b/go.mod
index 3c8dfe94e..83b62b8b0 100644
--- a/go.mod
+++ b/go.mod
@@ -33,7 +33,7 @@ require (
 	k8s.io/apimachinery v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20200729134348-d5654de09c73
-	knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf
+	knative.dev/eventing v0.20.1-0.20210217150902-78ee7893f583
 	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 	knative.dev/pkg v0.0.0-20210216013737-584933f8280b
 )
diff --git a/go.sum b/go.sum
index c71c8155a..550910574 100644
--- a/go.sum
+++ b/go.sum
@@ -1244,17 +1244,16 @@ k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6 h1:+WnxoVtG8TMiudHBSEtrVL
 k8s.io/kube-openapi v0.0.0-20200805222855-6aeccd4b50c6/go.mod h1:UuqjUnNftUyPE5H64/qeyjQoUZhGpeFDVdxjTeEVN2o=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K8Hf8whTseBgJcg=
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf h1:kjfXZLChfTzeXRDaXKmNO9EsnuVokkgzxPgikOJBfWM=
-knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf/go.mod h1:lxvfkunFc5wCcvlVqu4eJ3wbWqY4bTpQEDhNOMWD8+c=
+knative.dev/eventing v0.20.1-0.20210217150902-78ee7893f583 h1:i7bOF0RZ9b5pyc0rd3luR1aljPRBJ1ElGjolRIkPt0M=
+knative.dev/eventing v0.20.1-0.20210217150902-78ee7893f583/go.mod h1:3XibpVIriBFW31rn4uFQlTzqZgyKHgD4CTJHBwEPrUQ=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf h1:u4cY4jr2LYvhoz/1HBWEPsMiLkm0HMdDTfmmw1RE8zE=
 knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/pkg v0.0.0-20210208175252-a02dcff9ee26 h1:PkJB1M77wYBI1UF5vp8y4fRnYQDu1IkRWswp6CgRRSo=
-knative.dev/pkg v0.0.0-20210208175252-a02dcff9ee26/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f h1:pz+UDOTsCIClHXGM/SgCzkctLxgi3+P0C26x6Ujlqho=
 knative.dev/pkg v0.0.0-20210212203835-448ae657fb5f/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
+knative.dev/pkg v0.0.0-20210215165523-84c98f3c3e7a/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
 knative.dev/pkg v0.0.0-20210216013737-584933f8280b h1:2v+GBBenjPXhtk3KX/YqxCQVBU6o/AaRkkqE58lgFRQ=
 knative.dev/pkg v0.0.0-20210216013737-584933f8280b/go.mod h1:TJSdebQOWX5N2bszohOYVi0H1QtXbtlYLuMghAFBMhY=
-knative.dev/reconciler-test v0.0.0-20210214052044-2a6803e31481/go.mod h1:hIOndGzuU1mxSJ8IF0RqPL/rAoazZZ0NjuJoMwKeNBM=
+knative.dev/reconciler-test v0.0.0-20210215155505-fe22009753f9/go.mod h1:vB3GnWUdlkvtmoVUP9anmh82S1QC/LJKtOrAMA1COtw=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/apis/sources/v1beta1/kafka_validation.go b/pkg/apis/sources/v1beta1/kafka_validation.go
index 573fd321b..c68a20e7c 100644
--- a/pkg/apis/sources/v1beta1/kafka_validation.go
+++ b/pkg/apis/sources/v1beta1/kafka_validation.go
@@ -26,25 +26,58 @@ import (
 )
 
 // Validate ensures KafkaSource is properly configured.
-func (r *KafkaSource) Validate(ctx context.Context) *apis.FieldError {
+func (ks *KafkaSource) Validate(ctx context.Context) *apis.FieldError {
+	errs := ks.Spec.Validate(ctx).ViaField("spec")
 	if apis.IsInUpdate(ctx) {
 		original := apis.GetBaseline(ctx).(*KafkaSource)
-		// Ignore diffs in bootstrapServers and topics
-		ignoreArgs := cmpopts.IgnoreFields(KafkaSourceSpec{}, "Topics", "BootstrapServers")
-		if diff, err := kmp.ShortDiff(original.Spec, r.Spec, ignoreArgs); err != nil {
-			return &apis.FieldError{
-				Message: "Failed to diff KafkaSource",
-				Paths:   []string{"spec"},
-				Details: err.Error(),
-			}
-		} else if diff != "" {
-			return &apis.FieldError{
+		errs = errs.Also(ks.CheckImmutableFields(ctx, original))
+	}
+	return errs
+}
+
+func (kss *KafkaSourceSpec) Validate(ctx context.Context) *apis.FieldError {
+	var errs *apis.FieldError
+
+	// Validate sink
+	errs = errs.Also(kss.Sink.Validate(ctx).ViaField("sink"))
+
+	// We want to ensure we have a bootstrapServer & Topics for creation
+	if apis.IsInCreate(ctx) {
+		if len(kss.Topics) <= 0 {
+			errs = errs.Also(apis.ErrMissingField("Topics"))
+		}
+		if len(kss.BootstrapServers) <= 0 {
+			errs = errs.Also(apis.ErrMissingField("bootstrapServer"))
+		}
+	}
+
+	return errs
+}
+
+func (ks *KafkaSource) CheckImmutableFields(ctx context.Context, original *KafkaSource) *apis.FieldError {
+	if original == nil {
+		return nil
+	}
+	var errs *apis.FieldError
+
+	// Ignore diffs in bootstrapServers and topics
+	ignoreArgs := cmpopts.IgnoreFields(KafkaSourceSpec{}, "Topics", "BootstrapServers", "Sink", "Net", "CloudEventOverrides")
+	if diff, err := kmp.ShortDiff(original.Spec, ks.Spec, ignoreArgs); err != nil {
+		errs = errs.Also(&apis.FieldError{
+			Message: "Failed to diff KafkaSource",
+			Paths:   []string{"spec"},
+			Details: err.Error(),
+		})
+		return errs
+	} else if diff != "" {
+		errs = errs.Also(
+			&apis.FieldError{
 				Message: "Immutable fields changed (-old +new)",
 				Paths:   []string{"spec"},
 				Details: diff,
-			}
-		}
+			})
+		return errs
 	}
 
-	return nil
+	return errs
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
index 8b808c988..353811c26 100644
--- a/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/consumer_group_watcher.go
@@ -145,7 +145,9 @@ func (w *WatcherImpl) Terminate() {
 
 	w.watchers = nil
 	w.cachedConsumerGroups = nil
-	w.done <- struct{}{}
+	if w.done != nil {
+		w.done <- struct{}{}
+	}
 }
 
 // TODO explore returning a channel instead of a taking callback
diff --git a/pkg/channel/consolidated/reconciler/controller/controller.go b/pkg/channel/consolidated/reconciler/controller/controller.go
index c19ee14a8..98cc260e0 100644
--- a/pkg/channel/consolidated/reconciler/controller/controller.go
+++ b/pkg/channel/consolidated/reconciler/controller/controller.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/kelseyhightower/envconfig"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"go.uber.org/zap"
 
@@ -87,12 +88,12 @@ func NewController(
 	impl := kafkaChannelReconciler.NewImpl(ctx, r)
 
 	// Get and Watch the Kakfa config map and dynamically update Kafka configuration.
-	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, "config-kafka", metav1.GetOptions{}); err == nil {
-		cmw.Watch("config-kafka", func(configMap *v1.ConfigMap) {
+	if _, err := kubeclient.Get(ctx).CoreV1().ConfigMaps(system.Namespace()).Get(ctx, constants.SettingsConfigMapName, metav1.GetOptions{}); err == nil {
+		cmw.Watch(constants.SettingsConfigMapName, func(configMap *v1.ConfigMap) {
 			r.updateKafkaConfig(ctx, configMap)
 		})
 	} else if !apierrors.IsNotFound(err) {
-		logging.FromContext(ctx).With(zap.Error(err)).Fatal("Error reading ConfigMap 'config-kafka'")
+		logging.FromContext(ctx).With(zap.Error(err)).Fatalf("Error reading ConfigMap '%s'", constants.SettingsConfigMapName)
 	}
 
 	logging.FromContext(ctx).Info("Setting up event handlers")
diff --git a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
index ff6f9340e..a48c6d1bb 100644
--- a/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/controller/kafkachannel.go
@@ -18,13 +18,13 @@ package controller
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"strings"
 	"time"
 
 	"knative.dev/eventing-kafka/pkg/channel/consolidated/kafka"
 	"knative.dev/eventing-kafka/pkg/common/client"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -158,7 +158,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, kc *v1beta1.KafkaChannel
 
 	if r.kafkaConfig == nil {
 		if r.kafkaConfigError == nil {
-			r.kafkaConfigError = errors.New("The config map 'config-kafka' does not exist")
+			r.kafkaConfigError = fmt.Errorf("the config map '%s' does not exist", constants.SettingsConfigMapName)
 		}
 		kc.Status.MarkConfigFailed("MissingConfiguration", "%v", r.kafkaConfigError)
 		return r.kafkaConfigError
@@ -600,7 +600,9 @@ func (r *Reconciler) updateKafkaConfig(ctx context.Context, configMap *corev1.Co
 	}
 
 	if r.consumerGroupWatcher != nil {
+		logger.Info("terminating consumer group watcher")
 		r.consumerGroupWatcher.Terminate()
+		logger.Info("terminated consumer group watcher")
 	}
 
 	r.consumerGroupWatcher = NewConsumerGroupWatcher(ctx, ac, pollInterval)
@@ -616,6 +618,8 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, kc *v1beta1.KafkaChannel)
 			return err
 		}
 	}
-	r.consumerGroupWatcher.Forget(string(kc.ObjectMeta.UID))
+	if r.consumerGroupWatcher != nil {
+		r.consumerGroupWatcher.Forget(string(kc.ObjectMeta.UID))
+	}
 	return newReconciledNormal(kc.Namespace, kc.Name) //ok to remove finalizer
 }
diff --git a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
index 63fdfba7e..ac34d20be 100644
--- a/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
+++ b/pkg/channel/consolidated/reconciler/controller/resources/dispatcher.go
@@ -20,6 +20,7 @@ import (
 	v1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/system"
 )
 
@@ -77,7 +78,7 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 							VolumeMounts: []corev1.VolumeMount{
 								{
 									Name:      "config-kafka",
-									MountPath: "/etc/config-kafka",
+									MountPath: constants.SettingsConfigMapMountPath,
 								},
 							},
 						},
@@ -88,7 +89,7 @@ func MakeDispatcher(args DispatcherArgs) *v1.Deployment {
 							VolumeSource: corev1.VolumeSource{
 								ConfigMap: &corev1.ConfigMapVolumeSource{
 									LocalObjectReference: corev1.LocalObjectReference{
-										Name: "config-kafka",
+										Name: constants.SettingsConfigMapName,
 									},
 								},
 							},
diff --git a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
index e1e01a38d..87e0608ef 100644
--- a/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
+++ b/pkg/channel/consolidated/reconciler/dispatcher/kafkachannel.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
+	"knative.dev/eventing-kafka/pkg/common/constants"
 
 	"knative.dev/eventing/pkg/apis/eventing"
 	"knative.dev/eventing/pkg/channel/fanout"
@@ -76,7 +77,7 @@ func NewController(ctx context.Context, cmw configmap.Watcher) *controller.Impl
 		logger.Fatalw("unable to setup tracing", zap.Error(err))
 	}
 
-	configMap, err := configmap.Load("/etc/config-kafka")
+	configMap, err := configmap.Load(constants.SettingsConfigMapMountPath)
 	if err != nil {
 		logger.Fatalw("error loading configuration", zap.Error(err))
 	}
diff --git a/pkg/channel/distributed/common/config/configwatcher.go b/pkg/channel/distributed/common/config/configwatcher.go
index fd9e43c32..c8a64fb27 100644
--- a/pkg/channel/distributed/common/config/configwatcher.go
+++ b/pkg/channel/distributed/common/config/configwatcher.go
@@ -62,6 +62,7 @@ type EKKafkaTopicConfig struct {
 
 // EKKafkaConfig contains items relevant to Kafka specifically, and the Sarama logging flag
 type EKKafkaConfig struct {
+	Brokers             string             `json:"brokers,omitempty"`
 	EnableSaramaLogging bool               `json:"enableSaramaLogging,omitempty"`
 	Topic               EKKafkaTopicConfig `json:"topic,omitempty"`
 	AdminType           string             `json:"adminType,omitempty"`
diff --git a/pkg/channel/distributed/common/config/secretwatcher.go b/pkg/channel/distributed/common/config/secretwatcher.go
new file mode 100644
index 000000000..29e32af47
--- /dev/null
+++ b/pkg/channel/distributed/common/config/secretwatcher.go
@@ -0,0 +1,100 @@
+/*
+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 config
+
+import (
+	"context"
+	"time"
+
+	"k8s.io/apimachinery/pkg/fields"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/watch"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+	"knative.dev/eventing-kafka/pkg/common/client"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/logging"
+)
+
+type SecretObserver func(ctx context.Context, secret *corev1.Secret)
+
+//
+// Initialize The Specified Context With A Secret Watcher
+//
+func InitializeSecretWatcher(ctx context.Context, namespace string, name string, observer SecretObserver) error {
+
+	logger := logging.FromContext(ctx)
+	secrets := kubeclient.Get(ctx).CoreV1().Secrets(namespace)
+	watcher, err := secrets.Watch(ctx, metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("metadata.name", name).String()})
+	if err != nil {
+		logger.Error("Failed to start secret watcher", zap.Error(err))
+	}
+
+	go func() {
+		defer watcher.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				logger.Info("Stopped Secret Watcher")
+				return
+			case event := <-watcher.ResultChan():
+				// We only care if the secret was modified or added
+				if event.Type == watch.Added || event.Type == watch.Modified {
+					if secret, ok := event.Object.(*corev1.Secret); ok {
+						observer(ctx, secret)
+					} else {
+						logger.Error("Could not convert watched object to Secret", zap.Any("event.Object", event.Object))
+					}
+				}
+			case <-time.After(5 * time.Second):
+				// Don't hang a shutdown waiting for an event
+			}
+		}
+
+	}()
+
+	return nil
+}
+
+// Look Up And Return Kafka Auth ConfigAnd Brokers From Named Secret
+func GetAuthConfigFromKubernetes(ctx context.Context, secretName string, secretNamespace string) (*client.KafkaAuthConfig, error) {
+	secrets := kubeclient.Get(ctx).CoreV1().Secrets(secretNamespace)
+	secret, err := secrets.Get(ctx, secretName, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
+	}
+	kafkaAuthCfg := GetAuthConfigFromSecret(secret)
+	return kafkaAuthCfg, nil
+}
+
+// Look Up And Return Kafka Auth Config And Brokers From Provided Secret
+func GetAuthConfigFromSecret(secret *corev1.Secret) *client.KafkaAuthConfig {
+	if secret == nil || secret.Data == nil {
+		return nil
+	}
+
+	return &client.KafkaAuthConfig{
+		SASL: &client.KafkaSaslConfig{
+			User:     string(secret.Data[constants.KafkaSecretKeyUsername]),
+			Password: string(secret.Data[constants.KafkaSecretKeyPassword]),
+			SaslType: string(secret.Data[constants.KafkaSecretKeySaslType]),
+		},
+	}
+}
diff --git a/pkg/channel/distributed/common/env/constants.go b/pkg/channel/distributed/common/env/constants.go
index 58a615ff8..5cda38b9a 100644
--- a/pkg/channel/distributed/common/env/constants.go
+++ b/pkg/channel/distributed/common/env/constants.go
@@ -29,10 +29,8 @@ const (
 	ResyncPeriodMinutesEnvVarKey = "RESYNC_PERIOD_MINUTES"
 
 	// Kafka Authorization
-	KafkaBrokerEnvVarKey   = "KAFKA_BROKERS"
-	KafkaUsernameEnvVarKey = "KAFKA_USERNAME"
-	KafkaPasswordEnvVarKey = "KAFKA_PASSWORD"
-	KafkaSaslTypeEnvVarKey = "KAKFA_SASL_TYPE"
+	KafkaSecretNamespaceEnvVarKey = "KAFKA_SECRET_NAMESPACE"
+	KafkaSecretNameEnvVarKey      = "KAFKA_SECRET_NAME"
 
 	// 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 15a5d6589..644ace9f4 100644
--- a/pkg/channel/distributed/common/kafka/constants/constants.go
+++ b/pkg/channel/distributed/common/kafka/constants/constants.go
@@ -29,15 +29,11 @@ const (
 	KafkaSecretLabel = "eventing-kafka.knative.dev/kafka-secret"
 
 	// Kafka Secret Keys
-	KafkaSecretKeyBrokers   = "brokers"
 	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.
-
 	// Kafka Topic Config Keys
 	TopicDetailConfigRetentionMs = "retention.ms"
 
@@ -47,9 +43,6 @@ const (
 	EventHubErrorCodeCapacityLimit = 403
 	EventHubErrorCodeConflict      = 409
 
-	// EventHub Constraints
-	MaxEventHubNamespaces = 100
-
 	// KafkaChannel Constants
 	KafkaChannelServiceNameSuffix = "kn-channel" // Specific Value For Use With Knative e2e Tests!
 )
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama.go b/pkg/channel/distributed/common/kafka/sarama/sarama.go
index 4328e7ee0..08f5ac1c3 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama.go
@@ -68,6 +68,12 @@ func LoadSettings(ctx context.Context, clientId string, kafkaAuthConfig *client.
 	// Merge The ConfigMap Settings Into The Provided Config
 	saramaSettingsYamlString := configMap.Data[constants.SaramaSettingsConfigKey]
 
+	if kafkaAuthConfig != nil && kafkaAuthConfig.SASL.User == "" {
+		// The config builder expects the entire config object to be nil if not using auth
+		// (Otherwise it will end up with "PLAIN" SASL by default and fail due to having no user/password)
+		kafkaAuthConfig = nil
+	}
+
 	// Merge The Sarama Settings In The ConfigMap Into A New Base Sarama Config
 	saramaConfig, err := client.NewConfigBuilder().
 		WithDefaults().
@@ -94,3 +100,21 @@ func LoadEventingKafkaSettings(configMap *corev1.ConfigMap) (*commonconfig.Event
 
 	return eventingKafkaConfig, nil
 }
+
+// AuthFromSarama creates a KafkaAuthConfig using the SASL settings from
+// a given Sarama config, or nil if there is no SASL user in that config
+func AuthFromSarama(config *sarama.Config) *client.KafkaAuthConfig {
+	// Use the SASL settings from the provided Sarama config only if the user is non-empty
+	if config.Net.SASL.User != "" {
+		return &client.KafkaAuthConfig{
+			SASL: &client.KafkaSaslConfig{
+				User:     config.Net.SASL.User,
+				Password: config.Net.SASL.Password,
+				SaslType: string(config.Net.SASL.Mechanism),
+			},
+		}
+	} else {
+		// If the user is empty, return explicitly nil authentication
+		return nil
+	}
+}
diff --git a/pkg/channel/distributed/controller/constants/constants.go b/pkg/channel/distributed/controller/constants/constants.go
index dc7297f0a..8dfa42201 100644
--- a/pkg/channel/distributed/controller/constants/constants.go
+++ b/pkg/channel/distributed/controller/constants/constants.go
@@ -47,12 +47,6 @@ const (
 	// Refer to: https://knative.dev/eventing-kafka/blob/master/cmd/channel/main.go
 	HttpContainerPortNumber = 8080
 
-	// Kafka Secret Data Keys
-	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 8d71ec04f..c68f25fb1 100644
--- a/pkg/channel/distributed/controller/kafkachannel/controller.go
+++ b/pkg/channel/distributed/controller/kafkachannel/controller.go
@@ -72,15 +72,12 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 		logger.Fatal("Failed To Load Kafka Auth Secret", zap.Error(err))
 	} else if kafkaSecret == nil {
 		logger.Fatal("No Kafka Auth Secret Found")
-	} else if !util.ValidateKafkaSecret(logger, kafkaSecret) {
-		logger.Fatal("Found Invalid Kafka Auth Secret")
 	} else {
 		logger.Info("Found Valid Kafka Auth Secret")
 	}
 
 	// Extract The Relevant Data From The Kafka Secret
 	kafkaSecretName := kafkaSecret.Name
-	kafkaBrokers := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyBrokers])
 	kafkaUsername := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyUsername])
 	kafkaPassword := string(kafkaSecret.Data[clientconstants.KafkaSecretKeyPassword])
 	kafkaSaslType := string(kafkaSecret.Data[clientconstants.KafkaSecretKeySaslType])
@@ -136,7 +133,7 @@ func NewController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 		adminMutex:           &sync.Mutex{},
 		configObserver:       rec.configMapObserver, // Maintains a reference so that the ConfigWatcher can call it
 		kafkaSecret:          kafkaSecretName,
-		kafkaBrokers:         kafkaBrokers,
+		kafkaBrokers:         configuration.Kafka.Brokers,
 		kafkaUsername:        kafkaUsername,
 		kafkaPassword:        kafkaPassword,
 		kafkaSaslType:        kafkaSaslType,
diff --git a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
index bfc752201..1d4108fa0 100644
--- a/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
+++ b/pkg/channel/distributed/controller/kafkachannel/dispatcher.go
@@ -538,62 +538,26 @@ func (r *Reconciler) dispatcherDeploymentEnvVars(channel *kafkav1beta1.KafkaChan
 		},
 	}
 
-	// Get The Kafka Secret From The Reconciler
-	kafkaSecret := r.kafkaSecret
-
 	// If The Kafka Secret Env Var Is Specified Then Append Relevant Env Vars
-	if len(kafkaSecret) <= 0 {
+	if len(r.kafkaSecret) <= 0 {
 
 		// Received Invalid Kafka Secret - Cannot Proceed
 		return nil, fmt.Errorf("invalid kafkaSecret for topic '%s'", topicName)
 
 	} else {
 
-		// Append The Kafka Brokers As Env Var
-		envVars = append(envVars, corev1.EnvVar{
-			Name: commonenv.KafkaBrokerEnvVarKey,
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
-					Key:                  constants.KafkaSecretDataKeyBrokers,
-				},
-			},
-		})
-
-		// Append The Kafka Username As Env Var
+		// Append The Secret Name As Env Var
 		envVars = append(envVars, corev1.EnvVar{
-			Name: commonenv.KafkaUsernameEnvVarKey,
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
-					Key:                  constants.KafkaSecretDataKeyUsername,
-				},
-			},
+			Name:  commonenv.KafkaSecretNameEnvVarKey,
+			Value: r.kafkaSecret,
 		})
 
-		// Append The Kafka Password As Env Var
+		// Append The Secret Namespace As Env Var
 		envVars = append(envVars, corev1.EnvVar{
-			Name: commonenv.KafkaPasswordEnvVarKey,
-			ValueFrom: &corev1.EnvVarSource{
-				SecretKeyRef: &corev1.SecretKeySelector{
-					LocalObjectReference: corev1.LocalObjectReference{Name: kafkaSecret},
-					Key:                  constants.KafkaSecretDataKeyPassword,
-				},
-			},
+			Name:  commonenv.KafkaSecretNamespaceEnvVarKey,
+			Value: r.environment.SystemNamespace,
 		})
 
-		// 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/kafkasecret/receiver.go b/pkg/channel/distributed/controller/kafkasecret/receiver.go
index ebb581241..9af1e3416 100644
--- a/pkg/channel/distributed/controller/kafkasecret/receiver.go
+++ b/pkg/channel/distributed/controller/kafkasecret/receiver.go
@@ -409,49 +409,16 @@ func (r *Reconciler) receiverDeploymentEnvVars(secret *corev1.Secret) ([]corev1.
 		},
 	}
 
-	// Append The Kafka Brokers As Env Var
+	// Append The Secret Name As Env Var
 	envVars = append(envVars, corev1.EnvVar{
-		Name: commonenv.KafkaBrokerEnvVarKey,
-		ValueFrom: &corev1.EnvVarSource{
-			SecretKeyRef: &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
-				Key:                  constants.KafkaSecretDataKeyBrokers,
-			},
-		},
-	})
-
-	// Append The Kafka Username As Env Var
-	envVars = append(envVars, corev1.EnvVar{
-		Name: commonenv.KafkaUsernameEnvVarKey,
-		ValueFrom: &corev1.EnvVarSource{
-			SecretKeyRef: &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
-				Key:                  constants.KafkaSecretDataKeyUsername,
-			},
-		},
+		Name:  commonenv.KafkaSecretNameEnvVarKey,
+		Value: secret.Name,
 	})
 
-	// Append The Kafka Password As Env Var
+	// Append The Secret Namespace As Env Var
 	envVars = append(envVars, corev1.EnvVar{
-		Name: commonenv.KafkaPasswordEnvVarKey,
-		ValueFrom: &corev1.EnvVarSource{
-			SecretKeyRef: &corev1.SecretKeySelector{
-				LocalObjectReference: corev1.LocalObjectReference{Name: secret.Name},
-				Key:                  constants.KafkaSecretDataKeyPassword,
-			},
-		},
-	})
-
-	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,
-			},
-		},
+		Name:  commonenv.KafkaSecretNamespaceEnvVarKey,
+		Value: r.environment.SystemNamespace,
 	})
 
 	// Return The Receiver Deployment EnvVars Array
diff --git a/pkg/channel/distributed/controller/util/secret.go b/pkg/channel/distributed/controller/util/secret.go
index 77cb2ffa2..23d49b428 100644
--- a/pkg/channel/distributed/controller/util/secret.go
+++ b/pkg/channel/distributed/controller/util/secret.go
@@ -83,32 +83,3 @@ func GetKafkaSecret(ctx context.Context, k8sClient kubernetes.Interface, k8sName
 		return &secretList.Items[0], nil
 	}
 }
-
-// Utility Function For Validating Kafka Secret
-func ValidateKafkaSecret(logger *zap.Logger, secret *corev1.Secret) bool {
-
-	// Assume Invalid Until Proven Otherwise
-	valid := false
-
-	// Validate The Kafka Secret
-	if secret != nil {
-
-		// Extract The Relevant Data From The Kafka Secret
-		brokers := string(secret.Data[clientconstants.KafkaSecretKeyBrokers])
-
-		// Validate Kafka Secret Data
-		if len(brokers) > 0 {
-
-			// Mark Kafka Secret As Valid
-			valid = true
-
-		} else {
-
-			// Invalid Kafka Secret - Log State
-			logger.Error("Invalid Kafka Secret - Missing Brokers", zap.String("Secret", fmt.Sprintf("%s/%s", secret.Namespace, secret.Name)))
-		}
-	}
-
-	// Return Kafka Secret Validity
-	return valid
-}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index f0be4365a..3453e1a6a 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -19,8 +19,11 @@ package dispatcher
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
+
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
@@ -64,6 +67,7 @@ func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId st
 //  Dispatcher Interface
 type Dispatcher interface {
 	ConfigChanged(ctx context.Context, configMap *corev1.ConfigMap) Dispatcher
+	SecretChanged(ctx context.Context, secret *corev1.Secret) Dispatcher
 	Shutdown()
 	UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
 }
@@ -273,6 +277,12 @@ func (d *DispatcherImpl) ConfigChanged(ctx context.Context, configMap *corev1.Co
 
 	d.Logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
 
+	// Validate Configuration (Should Always Be Present)
+	if d.SaramaConfig == nil {
+		d.Logger.Error("Dispatcher configuration is nil; cannot reconfigure")
+		return nil
+	}
+
 	// Validate The ConfigMap Data
 	if configMap.Data == nil {
 		d.Logger.Error("Attempted to merge sarama settings with empty configmap")
@@ -287,22 +297,10 @@ func (d *DispatcherImpl) ConfigChanged(ctx context.Context, configMap *corev1.Co
 		WithDefaults().
 		FromYaml(saramaSettingsYamlString)
 
-	// Validate Configuration (Should Always Be Present)
-	if d.SaramaConfig != nil {
-		configBuilder = configBuilder.WithClientId(d.SaramaConfig.ClientID)
-
-		// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
-		if d.SaramaConfig.Net.SASL.User != "" {
-			kafkaAuthCfg := &client.KafkaAuthConfig{
-				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)
-		}
-	}
+	// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
+	configBuilder = configBuilder.
+		WithClientId(d.SaramaConfig.ClientID).
+		WithAuth(kafkasarama.AuthFromSarama(d.SaramaConfig))
 
 	newConfig, err := configBuilder.Build(ctx)
 	if err != nil {
@@ -310,27 +308,71 @@ func (d *DispatcherImpl) ConfigChanged(ctx context.Context, configMap *corev1.Co
 		return nil
 	}
 
-	// Validate Configuration (Should Always Be Present)
-	if d.SaramaConfig != nil {
-		// Enable Sarama Logging If Specified In ConfigMap
-		if ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap); err == nil && ekConfig != nil {
-			kafkasarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
-			d.Logger.Debug("Updated Sarama logging", zap.Bool("Kafka.EnableSaramaLogging", ekConfig.Kafka.EnableSaramaLogging))
-		} else {
-			d.Logger.Error("Could Not Extract Eventing-Kafka Setting From Updated ConfigMap", zap.Error(err))
-		}
+	ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap)
+	if err != nil || ekConfig == nil {
+		d.Logger.Error("Could Not Extract Eventing-Kafka Setting From Updated ConfigMap", zap.Error(err))
+		return nil
+	}
 
-		// Ignore the "Producer" section as changes to that do not require recreating the Dispatcher
-		if client.ConfigEqual(newConfig, d.SaramaConfig, newConfig.Producer) {
-			d.Logger.Info("No Consumer Changes Detected In New Configuration - Ignoring")
-			return nil
-		}
+	// Enable Sarama Logging If Specified In ConfigMap
+	kafkasarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
+	d.Logger.Debug("Updated Sarama logging", zap.Bool("Kafka.EnableSaramaLogging", ekConfig.Kafka.EnableSaramaLogging))
+
+	// Ignore the "Producer" section as changes to that do not require recreating the Dispatcher
+	if client.ConfigEqual(newConfig, d.SaramaConfig, newConfig.Producer) &&
+		client.HasSameBrokers(ekConfig.Kafka.Brokers, d.Brokers) {
+		d.Logger.Info("No Consumer Changes Detected In New Configuration - Ignoring")
+		return nil
 	}
 
 	// Create A New Dispatcher With The New Configuration (Reusing All Other Existing Config)
-	d.Logger.Info("Consumer Changes Detected In New Configuration - Recreating Dispatcher")
+	d.Logger.Info("Consumer Changes Detected In New Configuration - Closing & Recreating Consumer Groups")
+	return d.reconfigure(newConfig, ekConfig)
+}
+
+// SecretChanged is called by the secretObserver handler function in main() so that
+// settings specific to the dispatcher may be extracted and the dispatcher restarted if necessary.
+func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secret) Dispatcher {
+
+	// Debug Log The Secret Change
+	d.Logger.Debug("New Secret Received", zap.String("secret.Name", secret.ObjectMeta.Name))
+
+	kafkaAuthCfg := config.GetAuthConfigFromSecret(secret)
+	if kafkaAuthCfg == nil {
+		d.Logger.Warn("No auth config found in secret; ignoring update")
+		return nil
+	}
+
+	// Don't Restart Dispatcher If All Auth Settings Identical
+	if kafkaAuthCfg.SASL.HasSameSettings(d.SaramaConfig) {
+		d.Logger.Info("No relevant changes in Secret; ignoring update")
+		return nil
+	}
+
+	// Build New Config Using Existing Config And New Auth Settings
+	if kafkaAuthCfg.SASL.User == "" {
+		// The config builder expects the entire config object to be nil if not using auth
+		kafkaAuthCfg = nil
+	}
+	newConfig, err := client.NewConfigBuilder().WithExisting(d.SaramaConfig).WithAuth(kafkaAuthCfg).Build(ctx)
+	if err != nil {
+		d.Logger.Error("Unable to merge new auth into sarama settings", zap.Error(err))
+		return nil
+	}
+
+	// Create A New Dispatcher With The New Configuration (Reusing All Other Existing Config)
+	d.Logger.Info("Changes Detected In New Secret - Closing & Recreating Consumer Groups")
+	return d.reconfigure(newConfig, nil)
+}
+
+// Shut down the current dispatcher and recreate it with new settings
+func (d *DispatcherImpl) reconfigure(newConfig *sarama.Config, ekConfig *config.EventingKafkaConfig) Dispatcher {
 	d.Shutdown()
 	d.DispatcherConfig.SaramaConfig = newConfig
+	if ekConfig != nil {
+		// Currently the only thing that a new dispatcher might care about in the EventingKafkaConfig is the Brokers
+		d.DispatcherConfig.Brokers = strings.Split(ekConfig.Kafka.Brokers, ",")
+	}
 	newDispatcher := NewDispatcher(d.DispatcherConfig)
 	failedSubscriptions := newDispatcher.UpdateSubscriptions(d.SubscriberSpecs)
 	if len(failedSubscriptions) > 0 {
diff --git a/pkg/channel/distributed/dispatcher/env/environment.go b/pkg/channel/distributed/dispatcher/env/environment.go
index 4071afabb..0497b5909 100644
--- a/pkg/channel/distributed/dispatcher/env/environment.go
+++ b/pkg/channel/distributed/dispatcher/env/environment.go
@@ -45,16 +45,14 @@ type Environment struct {
 	HealthPort int // Required
 
 	// Kafka Configuration
-	KafkaBrokers string        // Required
 	KafkaTopic   string        // Required
 	ChannelKey   string        // Required
 	ServiceName  string        // Required
 	ResyncPeriod time.Duration // Optional
 
 	// Kafka Authorization
-	KafkaUsername string // Optional
-	KafkaPassword string // Optional
-	KafkaSaslType string // Optional
+	KafkaSecretName      string // Required
+	KafkaSecretNamespace string // Required
 }
 
 // Get The Environment
@@ -102,8 +100,14 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		return nil, err
 	}
 
-	// Get The Required K8S KafkaBrokers Config Value
-	environment.KafkaBrokers, err = env.GetRequiredConfigValue(logger, env.KafkaBrokerEnvVarKey)
+	// Get The Required K8S KafkaSecretName Config Value
+	environment.KafkaSecretName, err = env.GetRequiredConfigValue(logger, env.KafkaSecretNameEnvVarKey)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get The Required K8S KafkaSecretNamespace Config Value
+	environment.KafkaSecretNamespace, err = env.GetRequiredConfigValue(logger, env.KafkaSecretNamespaceEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
@@ -137,23 +141,8 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 	environment.ResyncPeriod = time.Duration(resyncMinutes) * time.Minute
 
-	// Get The Optional KafkaUsername Config Value
-	environment.KafkaUsername = env.GetOptionalConfigValue(logger, env.KafkaUsernameEnvVarKey, "")
-
-	// 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 {
-		safeEnvironment.KafkaPassword = "*************"
-	}
-
 	// Log The Dispatcher Configuration Loaded From Environment Variables
-	logger.Info("Environment Variables", zap.Any("Environment", safeEnvironment))
+	logger.Info("Environment Variables", zap.Any("Environment", environment))
 
 	// Return The Populated Dispatcher Configuration Environment Structure
 	return environment, nil
diff --git a/pkg/channel/distributed/receiver/env/environment.go b/pkg/channel/distributed/receiver/env/environment.go
index a9f0a9975..1b7ce043b 100644
--- a/pkg/channel/distributed/receiver/env/environment.go
+++ b/pkg/channel/distributed/receiver/env/environment.go
@@ -45,14 +45,12 @@ type Environment struct {
 	HealthPort int // Required
 
 	// Kafka Configuration
-	KafkaBrokers string        // Required
 	ServiceName  string        // Required
 	ResyncPeriod time.Duration // Optional
 
 	// Kafka Authorization
-	KafkaUsername string // Optional
-	KafkaPassword string // Optional
-	KafkaSaslType string // Optional
+	KafkaSecretName      string // Required
+	KafkaSecretNamespace string // Required
 }
 
 // Get The Environment
@@ -100,8 +98,14 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 		return nil, err
 	}
 
-	// Get The Required K8S KafkaBrokers Config Value
-	environment.KafkaBrokers, err = env.GetRequiredConfigValue(logger, env.KafkaBrokerEnvVarKey)
+	// Get The Required K8S KafkaSecretNamespace Config Value
+	environment.KafkaSecretNamespace, err = env.GetRequiredConfigValue(logger, env.KafkaSecretNamespaceEnvVarKey)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get The Required K8S KafkaSecretName Config Value
+	environment.KafkaSecretName, err = env.GetRequiredConfigValue(logger, env.KafkaSecretNameEnvVarKey)
 	if err != nil {
 		return nil, err
 	}
@@ -123,23 +127,8 @@ func GetEnvironment(logger *zap.Logger) (*Environment, error) {
 	}
 	environment.ResyncPeriod = time.Duration(resyncMinutes) * time.Minute
 
-	// Get The Optional KafkaUsername Config Value
-	environment.KafkaUsername = env.GetOptionalConfigValue(logger, env.KafkaUsernameEnvVarKey, "")
-
-	// 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 {
-		safeEnvironment.KafkaPassword = "*************"
-	}
-
 	// Log The Receiver Configuration Loaded From Environment Variables
-	logger.Info("Environment Variables", zap.Any("Environment", safeEnvironment))
+	logger.Info("Environment Variables", zap.Any("Environment", environment))
 
 	// Return The Populated Environment Structure
 	return environment, nil
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index b18899785..6b9ffe764 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -19,6 +19,7 @@ package producer
 import (
 	"context"
 	"errors"
+	"strings"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -28,6 +29,7 @@ import (
 	"go.opencensus.io/trace"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	"knative.dev/eventing-kafka/pkg/channel/distributed/common/config"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/producer"
 	kafkasarama "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/sarama"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
@@ -71,7 +73,7 @@ func NewProducer(logger *zap.Logger,
 	}
 
 	// Create A New Producer
-	producer := &Producer{
+	newProducer := &Producer{
 		logger:             logger,
 		kafkaProducer:      kafkaProducer,
 		healthServer:       healthServer,
@@ -84,14 +86,14 @@ func NewProducer(logger *zap.Logger,
 	}
 
 	// Start Observing Metrics
-	producer.ObserveMetrics(constants.MetricsInterval)
+	newProducer.ObserveMetrics(constants.MetricsInterval)
 
 	// Mark The Producer As Ready
 	healthServer.SetProducerReady(true)
 
 	// Return The New Producer
 	logger.Info("Successfully Started Kafka Producer")
-	return producer, nil
+	return newProducer, nil
 }
 
 // Produce A KafkaMessage From The Specified CloudEvent To The Specified Topic And Wait For The Delivery Report
@@ -172,9 +174,14 @@ func (p *Producer) ObserveMetrics(interval time.Duration) {
 // then just log that and move on; do not restart the Producer unnecessarily.
 func (p *Producer) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMap) *Producer {
 
-	// Debug Log The ConfigMap Change
 	p.logger.Debug("New ConfigMap Received", zap.String("configMap.Name", configMap.ObjectMeta.Name))
 
+	// Validate Configuration (Should Always Be Present)
+	if p.configuration == nil {
+		p.logger.Error("Producer configuration is nil; cannot reconfigure")
+		return nil
+	}
+
 	// Validate The ConfigMap Data
 	if configMap.Data == nil {
 		p.logger.Error("Attempted to merge sarama settings with empty configmap")
@@ -189,21 +196,10 @@ func (p *Producer) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMa
 		WithDefaults().
 		FromYaml(saramaSettingsYamlString)
 
-	if p.configuration != nil {
-		configBuilder = configBuilder.WithClientId(p.configuration.ClientID)
-
-		// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
-		if p.configuration.Net.SASL.User != "" {
-			kafkaAuthCfg := &client.KafkaAuthConfig{
-				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)
-		}
-	}
+	// Some of the current config settings may not be overridden by the configmap (username, password, etc.)
+	configBuilder = configBuilder.
+		WithClientId(p.configuration.ClientID).
+		WithAuth(kafkasarama.AuthFromSarama(p.configuration))
 
 	newConfig, err := configBuilder.Build(ctx)
 	if err != nil {
@@ -211,35 +207,61 @@ func (p *Producer) ConfigChanged(ctx context.Context, configMap *corev1.ConfigMa
 		return nil
 	}
 
-	// Validate Configuration (Should Always Be Present)
-	if p.configuration != nil {
-		// Enable Sarama Logging If Specified In ConfigMap
-		if ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap); err == nil && ekConfig != nil {
-			kafkasarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
-			p.logger.Debug("Updated Sarama logging", zap.Bool("Kafka.EnableSaramaLogging", ekConfig.Kafka.EnableSaramaLogging))
-		} else {
-			p.logger.Error("Could Not Extract Eventing-Kafka Setting From Updated ConfigMap", zap.Error(err))
-		}
+	ekConfig, err := kafkasarama.LoadEventingKafkaSettings(configMap)
+	if err != nil || ekConfig == nil {
+		p.logger.Error("Could Not Extract Eventing-Kafka Setting From Updated ConfigMap", zap.Error(err))
+		return nil
+	}
 
-		// Ignore the "Admin" and "Consumer" sections when comparing, as changes to those do not require restarting the Producer
-		if client.ConfigEqual(newConfig, p.configuration, newConfig.Admin, newConfig.Consumer) {
-			p.logger.Info("No Producer Changes Detected In New Configuration - Ignoring")
-			return nil
-		}
+	// Enable Sarama Logging If Specified In ConfigMap
+	kafkasarama.EnableSaramaLogging(ekConfig.Kafka.EnableSaramaLogging)
+	p.logger.Debug("Updated Sarama logging", zap.Bool("Kafka.EnableSaramaLogging", ekConfig.Kafka.EnableSaramaLogging))
+
+	// Ignore the "Admin" and "Consumer" sections when comparing, as changes to those do not require restarting the Producer
+	if client.ConfigEqual(newConfig, p.configuration, newConfig.Admin, newConfig.Consumer) &&
+		client.HasSameBrokers(ekConfig.Kafka.Brokers, p.brokers) {
+		p.logger.Info("No Producer Changes Detected In New Configuration - Ignoring")
+		return nil
 	}
 
 	// Create A New Producer With The New Configuration (Reusing All Other Existing Config)
 	p.logger.Info("Producer Changes Detected In New Configuration - Closing & Recreating Producer")
-	p.Close()
-	reconfiguredKafkaProducer, err := NewProducer(p.logger, newConfig, p.brokers, p.statsReporter, p.healthServer)
+	return p.reconfigure(newConfig, ekConfig)
+}
+
+// SecretChanged is called by the secretObserver handler function in main() so that
+// settings specific to the producer may be extracted and the producer restarted if necessary.
+func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Producer {
+
+	// Debug Log The Secret Change
+	p.logger.Debug("New Secret Received", zap.String("secret.Name", secret.ObjectMeta.Name))
+
+	kafkaAuthCfg := config.GetAuthConfigFromSecret(secret)
+	if kafkaAuthCfg == nil {
+		p.logger.Warn("No auth config found in secret; ignoring update")
+		return nil
+	}
+
+	// Don't Restart Producer If All Auth Settings Identical.
+	if kafkaAuthCfg.SASL.HasSameSettings(p.configuration) {
+		p.logger.Info("No relevant changes in Secret; ignoring update")
+		return nil
+	}
+
+	// Build New Config Using Existing Config And New Auth Settings
+	if kafkaAuthCfg.SASL.User == "" {
+		// The config builder expects the entire config object to be nil if not using auth
+		kafkaAuthCfg = nil
+	}
+	newConfig, err := client.NewConfigBuilder().WithExisting(p.configuration).WithAuth(kafkaAuthCfg).Build(ctx)
 	if err != nil {
-		p.logger.Fatal("Failed To Create Kafka Producer With New Configuration", zap.Error(err))
+		p.logger.Error("Unable to merge new auth into sarama settings", zap.Error(err))
 		return nil
 	}
 
-	// Successfully Created New Producer - Close Old One And Return New One
-	p.logger.Info("Successfully Created New Producer")
-	return reconfiguredKafkaProducer
+	// Create A New Producer With The New Configuration (Reusing All Other Existing Config)
+	p.logger.Info("Changes Detected In New Secret - Closing & Recreating Producer")
+	return p.reconfigure(newConfig, nil)
 }
 
 // Close The Producer (Stop Processing)
@@ -260,3 +282,21 @@ func (p *Producer) Close() {
 		p.logger.Info("Successfully Closed Kafka Producer")
 	}
 }
+
+// Shut down the current producer and recreate it with new settings
+func (p *Producer) reconfigure(newConfig *sarama.Config, ekConfig *config.EventingKafkaConfig) *Producer {
+	p.Close()
+	if ekConfig != nil {
+		// Currently the only thing that a new producer might care about in the EventingKafkaConfig is the Brokers
+		p.brokers = strings.Split(ekConfig.Kafka.Brokers, ",")
+	}
+	reconfiguredKafkaProducer, err := NewProducer(p.logger, newConfig, p.brokers, p.statsReporter, p.healthServer)
+	if err != nil {
+		p.logger.Fatal("Failed To Create Kafka Producer With New Configuration", zap.Error(err))
+		return nil
+	}
+
+	// Successfully Created New Producer - Close Old One And Return New One
+	p.logger.Info("Successfully Created New Producer")
+	return reconfiguredKafkaProducer
+}
diff --git a/pkg/common/client/config.go b/pkg/common/client/config.go
index c73f44833..340aa6471 100644
--- a/pkg/common/client/config.go
+++ b/pkg/common/client/config.go
@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -48,6 +49,41 @@ type KafkaSaslConfig struct {
 	SaslType string
 }
 
+// HasSameSettings returns true if all of the SASL settings in the provided config are the same as in this struct
+func (c *KafkaSaslConfig) HasSameSettings(saramaConfig *sarama.Config) bool {
+	return saramaConfig.Net.SASL.User == c.User &&
+		saramaConfig.Net.SASL.Password == c.Password &&
+		string(saramaConfig.Net.SASL.Mechanism) == c.SaslType
+}
+
+// HasSameBrokers returns true if all of the brokers in the slice are present and in the same order as
+// the ones in the consolidated str
+func HasSameBrokers(brokerString string, brokers []string) bool {
+
+	// Special case for empty string and slice, because Split() always returns at least one element
+	if brokerString == "" && len(brokers) == 0 {
+		return true
+	}
+
+	splitBrokers := strings.Split(brokerString, ",")
+
+	// Not the same if there aren't the same number of brokers
+	if len(splitBrokers) != len(brokers) {
+		return false
+	}
+
+	// Not the same if any of the individual brokers are different
+	// Note that brokers must be in the same order in the string and the slice in order to be considered equal
+	for index := range brokers {
+		if splitBrokers[index] != brokers[index] {
+			return false
+		}
+	}
+
+	// Same number and values of brokers
+	return true
+}
+
 // ConfigBuilder builds the Sarama config using multiple options.
 // Precedence for the options is following:
 // - get existing config if provided, create a new one otherwise
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 1b4272980..43c741803 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -23,8 +23,15 @@ const (
 	DefaultReplicationFactor = 1
 
 	// The name of the configmap used to hold eventing-kafka settings
-	SettingsConfigMapName          = "config-kafka"
+	SettingsConfigMapName = "config-kafka"
+	SettingsSecretName    = "kafka-cluster"
+
+	// Mount path of the configmap used to hold eventing-kafka settings
+	SettingsConfigMapMountPath = "/etc/" + SettingsConfigMapName
+
+	// Config key of the config in the configmap used to hold eventing-kafka settings
 	EventingKafkaSettingsConfigKey = "eventing-kafka"
+
 	// The name of the keys in the Data section of the eventing-kafka configmap that holds Sarama and Eventing-Kafka configuration YAML
 	SaramaSettingsConfigKey = "sarama"
 )
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/config/ping_defaults.go b/vendor/knative.dev/eventing/pkg/apis/sources/config/ping_defaults.go
new file mode 100644
index 000000000..77aa40ca8
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/config/ping_defaults.go
@@ -0,0 +1,87 @@
+/*
+Copyright 2020 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 config
+
+import (
+	"encoding/json"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	"sigs.k8s.io/yaml"
+)
+
+const (
+	// PingDefaultsConfigName is the name of config map for the default
+	// configs that pings should use.
+	PingDefaultsConfigName = "config-ping-webhook"
+
+	// PingDefaulterKey is the key in the ConfigMap to get the name of the default
+	// Ping CRD.
+	PingDefaulterKey = "ping-config"
+
+	PingDataMaxSize = -1
+)
+
+// NewPingDefaultsConfigFromMap creates a Defaults from the supplied Map
+func NewPingDefaultsConfigFromMap(data map[string]string) (*PingDefaults, error) {
+	nc := &PingDefaults{DataMaxSize: PingDataMaxSize}
+
+	// Parse out the Broker Configuration Cluster default section
+	value, present := data[PingDefaulterKey]
+	if !present || value == "" {
+		return nil, fmt.Errorf("ConfigMap is missing (or empty) key: %q : %v", PingDefaulterKey, data)
+	}
+	if err := parseEntry(value, nc); err != nil {
+		return nil, fmt.Errorf("Failed to parse the entry: %s", err)
+	}
+	return nc, nil
+}
+
+func parseEntry(entry string, out interface{}) error {
+	j, err := yaml.YAMLToJSON([]byte(entry))
+	if err != nil {
+		return fmt.Errorf("ConfigMap's value could not be converted to JSON: %s : %v", err, entry)
+	}
+	return json.Unmarshal(j, &out)
+}
+
+// NewPingDefaultsConfigFromConfigMap creates a PingDefaults from the supplied configMap
+func NewPingDefaultsConfigFromConfigMap(config *corev1.ConfigMap) (*PingDefaults, error) {
+	return NewPingDefaultsConfigFromMap(config.Data)
+}
+
+// PingDefaults includes the default values to be populated by the webhook.
+type PingDefaults struct {
+	DataMaxSize int `json:"dataMaxSize"`
+}
+
+func (d *PingDefaults) GetPingConfig() *PingDefaults {
+	if d.DataMaxSize < 0 {
+		d.DataMaxSize = PingDataMaxSize
+	}
+	return d
+
+}
+
+func (d *PingDefaults) DeepCopy() *PingDefaults {
+	if d == nil {
+		return nil
+	}
+	out := new(PingDefaults)
+	*out = *d
+	return out
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/config/store.go b/vendor/knative.dev/eventing/pkg/apis/sources/config/store.go
new file mode 100644
index 000000000..3b5a567d6
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/config/store.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2020 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 config
+
+import (
+	"context"
+
+	"knative.dev/pkg/configmap"
+)
+
+type pingCfgKey struct{}
+
+// Config holds the collection of configurations that we attach to contexts.
+// +k8s:deepcopy-gen=false
+type Config struct {
+	PingDefaults *PingDefaults
+}
+
+// FromContext extracts a Config from the provided context.
+func FromContext(ctx context.Context) *Config {
+	x, ok := ctx.Value(pingCfgKey{}).(*Config)
+	if ok {
+		return x
+	}
+	return nil
+}
+
+// FromContextOrDefaults is like FromContext, but when no Config is attached it
+// returns a Config populated with the defaults for each of the Config fields.
+func FromContextOrDefaults(ctx context.Context) *Config {
+	if cfg := FromContext(ctx); cfg != nil {
+		return cfg
+	}
+	pingDefaults, err := NewPingDefaultsConfigFromMap(map[string]string{})
+	if err != nil || pingDefaults == nil {
+		pingDefaults = &PingDefaults{DataMaxSize: PingDataMaxSize}
+		pingDefaults.GetPingConfig()
+	}
+
+	return &Config{
+		PingDefaults: pingDefaults,
+	}
+}
+
+// ToContext attaches the provided Config to the provided context, returning the
+// new context with the Config attached.
+func ToContext(ctx context.Context, c *Config) context.Context {
+	return context.WithValue(ctx, pingCfgKey{}, c)
+}
+
+// Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type Store struct {
+	*configmap.UntypedStore
+}
+
+// NewStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *Store {
+	store := &Store{
+		UntypedStore: configmap.NewUntypedStore(
+			"pingdefaults",
+			logger,
+			configmap.Constructors{
+				PingDefaultsConfigName: NewPingDefaultsConfigFromConfigMap,
+			},
+			onAfterStore...,
+		),
+	}
+
+	return store
+}
+
+// ToContext attaches the current Config state to the provided context.
+func (s *Store) ToContext(ctx context.Context) context.Context {
+	return ToContext(ctx, s.Load())
+}
+
+// Load creates a Config from the current config state of the Store.
+func (s *Store) Load() *Config {
+	return &Config{
+		PingDefaults: s.UntypedLoad(PingDefaultsConfigName).(*PingDefaults).DeepCopy(),
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/ping_validation.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/ping_validation.go
index ae7808482..8d0489494 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/ping_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1alpha2/ping_validation.go
@@ -18,11 +18,13 @@ package v1alpha2
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/robfig/cron/v3"
 	"knative.dev/pkg/apis"
 
 	"knative.dev/eventing/pkg/apis/eventing"
+	"knative.dev/eventing/pkg/apis/sources/config"
 )
 
 func (c *PingSource) Validate(ctx context.Context) *apis.FieldError {
@@ -38,6 +40,14 @@ func (cs *PingSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(fe)
 	}
 
+	pingConfig := config.FromContextOrDefaults(ctx)
+	pingDefaults := pingConfig.PingDefaults.GetPingConfig()
+
+	if bsize := len(cs.JsonData); pingDefaults.DataMaxSize > -1 && bsize > pingDefaults.DataMaxSize {
+		fe := apis.ErrInvalidValue(fmt.Sprintf("the jsonData length of %d bytes exceeds limit set at %d.", bsize, pingDefaults.DataMaxSize), "jsonData")
+		errs = errs.Also(fe)
+	}
+
 	if fe := cs.Sink.Validate(ctx); fe != nil {
 		errs = errs.Also(fe.ViaField("sink"))
 	}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/ping_validation.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/ping_validation.go
index e65ceacba..df07d799b 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/ping_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta1/ping_validation.go
@@ -18,10 +18,13 @@ package v1beta1
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/robfig/cron/v3"
 	"knative.dev/pkg/apis"
+
+	"knative.dev/eventing/pkg/apis/sources/config"
 )
 
 func (c *PingSource) Validate(ctx context.Context) *apis.FieldError {
@@ -46,6 +49,15 @@ func (cs *PingSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 		}
 	}
 
+	pingConfig := config.FromContextOrDefaults(ctx)
+
+	pingDefaults := pingConfig.PingDefaults.GetPingConfig()
+
+	if bsize := len(cs.JsonData); pingDefaults.DataMaxSize > -1 && bsize > pingDefaults.DataMaxSize {
+		fe := apis.ErrInvalidValue(fmt.Sprintf("the jsonData length of %d bytes exceeds limit set at %d.", bsize, pingDefaults.DataMaxSize), "jsonData")
+		errs = errs.Also(fe)
+	}
+
 	if fe := cs.Sink.Validate(ctx); fe != nil {
 		errs = errs.Also(fe.ViaField("sink"))
 	}
diff --git a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta2/ping_validation.go b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta2/ping_validation.go
index 325fe9136..189a38ccb 100644
--- a/vendor/knative.dev/eventing/pkg/apis/sources/v1beta2/ping_validation.go
+++ b/vendor/knative.dev/eventing/pkg/apis/sources/v1beta2/ping_validation.go
@@ -20,12 +20,15 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/json"
+	"fmt"
 	"strings"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 
 	"github.com/robfig/cron/v3"
 	"knative.dev/pkg/apis"
+
+	"knative.dev/eventing/pkg/apis/sources/config"
 )
 
 func (c *PingSource) Validate(ctx context.Context) *apis.FieldError {
@@ -50,6 +53,9 @@ func (cs *PingSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 		}
 	}
 
+	pingConfig := config.FromContextOrDefaults(ctx)
+	pingDefaults := pingConfig.PingDefaults.GetPingConfig()
+
 	if fe := cs.Sink.Validate(ctx); fe != nil {
 		errs = errs.Also(fe.ViaField("sink"))
 	}
@@ -57,6 +63,10 @@ func (cs *PingSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 	if cs.Data != "" && cs.DataBase64 != "" {
 		errs = errs.Also(apis.ErrMultipleOneOf("data", "dataBase64"))
 	} else if cs.DataBase64 != "" {
+		if bsize := len(cs.DataBase64); pingDefaults.DataMaxSize > -1 && bsize > pingDefaults.DataMaxSize {
+			fe := apis.ErrInvalidValue(fmt.Sprintf("the dataBase64 length of %d bytes exceeds limit set at %d.", bsize, pingDefaults.DataMaxSize), "dataBase64")
+			errs = errs.Also(fe)
+		}
 		decoded, err := base64.StdEncoding.DecodeString(cs.DataBase64)
 		// invalid base64 string
 		if err != nil {
@@ -69,13 +79,18 @@ func (cs *PingSourceSpec) Validate(ctx context.Context) *apis.FieldError {
 				}
 			}
 		}
-	} else if cs.Data != "" && cs.ContentType == cloudevents.ApplicationJSON {
-		// validate if data is valid JSON
-		if err := validateJSON(cs.Data); err != nil {
-			errs = errs.Also(apis.ErrInvalidValue(err, "data"))
+	} else if cs.Data != "" {
+		if bsize := len(cs.Data); pingDefaults.DataMaxSize > -1 && bsize > pingDefaults.DataMaxSize {
+			fe := apis.ErrInvalidValue(fmt.Sprintf("the data length of %d bytes exceeds limit set at %d.", bsize, pingDefaults.DataMaxSize), "data")
+			errs = errs.Also(fe)
+		}
+		if cs.ContentType == cloudevents.ApplicationJSON {
+			// validate if data is valid JSON
+			if err := validateJSON(cs.Data); err != nil {
+				errs = errs.Also(apis.ErrInvalidValue(err, "data"))
+			}
 		}
 	}
-
 	return errs
 }
 
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index 94f39494c..bb0127e55 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -129,7 +129,7 @@ func RetryConfigFromDeliverySpec(spec duckv1.DeliverySpec) (RetryConfig, error)
 
 	retryConfig := NoRetries()
 
-	retryConfig.CheckRetry = checkRetry
+	retryConfig.CheckRetry = RetryIfGreaterThan300
 
 	if spec.Retry != nil {
 		retryConfig.RetryMax = int(*spec.Retry)
@@ -160,6 +160,55 @@ func RetryConfigFromDeliverySpec(spec duckv1.DeliverySpec) (RetryConfig, error)
 	return retryConfig, nil
 }
 
-func checkRetry(_ context.Context, resp *nethttp.Response, err error) (bool, error) {
-	return !(resp != nil && resp.StatusCode < 300), err
+// Simple default implementation
+func RetryIfGreaterThan300(_ context.Context, response *nethttp.Response, err error) (bool, error) {
+	return !(response != nil && (response.StatusCode < 300)), err
+}
+
+// Alternative function to determine whether to retry based on response
+//
+// Note - Returning true indicates a retry should occur.  Returning an error will result in that
+//        error being returned instead of any errors from the Request.
+//
+// A retry is triggered for:
+// * nil responses
+// * emitted errors
+// * status codes that are 5XX, 404, 409, 429 as well if the statuscode is -1.
+func SelectiveRetry(_ context.Context, response *nethttp.Response, err error) (bool, error) {
+
+	// Retry Any Nil HTTP Response
+	if response == nil {
+		return true, nil
+	}
+
+	// Retry Any Errors
+	if err != nil {
+		return true, nil
+	}
+
+	// Extract The StatusCode From The Response & Add To Logger
+	statusCode := response.StatusCode
+
+	// Note - Normally we would NOT want to retry 4xx responses, BUT there are a few
+	//        known areas of knative-eventing that return codes in this range which
+	//        require retries.  Reasons for particular codes are as follows:
+	//
+	// 404  Although we would ideally not want to retry a permanent "Not Found"
+	//      response, a 404 can be returned when a pod is in the process of becoming
+	//      ready, so a retry can be a useful thing in this situation.
+	// 409  Returned by the E2E tests, so we must retry when "Conflict" is received, or the
+	//      tests will fail (see knative.dev/eventing/test/lib/recordevents/receiver/receiver.go)
+	// 429  Since retry typically involves a delay (usually an exponential backoff),
+	//      retrying after receiving a "Too Many Requests" response is useful.
+
+	if statusCode >= 500 || statusCode == 404 || statusCode == 429 || statusCode == 409 {
+		return true, nil
+	} else if statusCode >= 300 && statusCode <= 399 {
+		return false, nil
+	} else if statusCode == -1 {
+		return true, nil
+	}
+
+	// Do Not Retry 1XX, 2XX, 3XX & Most 4XX StatusCode Responses
+	return false, nil
 }
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 35d90e72e..6e32e6c92 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1080,7 +1080,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210215182711-24b1685ecddf
+# knative.dev/eventing v0.20.1-0.20210217150902-78ee7893f583
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1103,6 +1103,7 @@ knative.dev/eventing/pkg/apis/messaging/config
 knative.dev/eventing/pkg/apis/messaging/v1
 knative.dev/eventing/pkg/apis/messaging/v1beta1
 knative.dev/eventing/pkg/apis/sources
+knative.dev/eventing/pkg/apis/sources/config
 knative.dev/eventing/pkg/apis/sources/v1
 knative.dev/eventing/pkg/apis/sources/v1alpha1
 knative.dev/eventing/pkg/apis/sources/v1alpha2


Test patch:
diff --git a/pkg/apis/sources/v1beta1/kafka_validation_test.go b/pkg/apis/sources/v1beta1/kafka_validation_test.go
index 1c47cd5d7..aac512c87 100644
--- a/pkg/apis/sources/v1beta1/kafka_validation_test.go
+++ b/pkg/apis/sources/v1beta1/kafka_validation_test.go
@@ -29,6 +29,14 @@ var (
 	fullSpec = KafkaSourceSpec{
 		KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
 			BootstrapServers: []string{"servers"},
+			Net: bindingsv1beta1.KafkaNetSpec{
+				SASL: bindingsv1beta1.KafkaSASLSpec{
+					Enable: false,
+				},
+				TLS: bindingsv1beta1.KafkaTLSSpec{
+					Enable: false,
+				},
+			},
 		},
 		Topics:        []string{"topics"},
 		ConsumerGroup: "group",
@@ -44,6 +52,64 @@ var (
 	}
 )
 
+func TestKafkaSourceCheckRequiredFields(t *testing.T) {
+	testCases := map[string]struct {
+		orig    *KafkaSourceSpec
+		allowed bool
+	}{
+		"nil original": {
+			orig:    &KafkaSourceSpec{},
+			allowed: false,
+		},
+		"nil topic": {
+			allowed: false,
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				SourceSpec:    fullSpec.SourceSpec,
+			},
+		},
+		"nil bootstrapServer": {
+			orig: &KafkaSourceSpec{
+				Topics:     fullSpec.Topics,
+				SourceSpec: fullSpec.SourceSpec,
+			},
+			allowed: false,
+		},
+		"nil sink": {
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				Topics:        fullSpec.Topics,
+			},
+			allowed: false,
+		},
+		"min required fields": {
+			orig: &KafkaSourceSpec{
+				KafkaAuthSpec: fullSpec.KafkaAuthSpec,
+				Topics:        fullSpec.Topics,
+				SourceSpec:    fullSpec.SourceSpec,
+			},
+			allowed: true,
+		},
+		"full source": {
+			orig:    &fullSpec,
+			allowed: true,
+		},
+	}
+	for n, tc := range testCases {
+		t.Run(n, func(t *testing.T) {
+			ctx := context.TODO()
+			orig := &KafkaSource{
+				Spec: *tc.orig,
+			}
+			ctx = apis.WithinCreate(ctx)
+			err := orig.Validate(ctx)
+			if tc.allowed != (err == nil) {
+				t.Fatalf("Required Field Not Included: %v", err)
+			}
+		})
+	}
+}
+
 func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 	testCases := map[string]struct {
 		orig    *KafkaSourceSpec
@@ -77,7 +143,8 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 		"Sink.APIVersion changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -89,12 +156,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Kind changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -106,12 +174,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Namespace changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -123,12 +192,13 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"Sink.Name changed": {
 			orig: &fullSpec,
 			updated: KafkaSourceSpec{
-				Topics: fullSpec.Topics,
+				Topics:        fullSpec.Topics,
+				ConsumerGroup: fullSpec.ConsumerGroup,
 				SourceSpec: duckv1.SourceSpec{
 					Sink: duckv1.Destination{
 						Ref: &duckv1.KReference{
@@ -140,7 +210,7 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 					},
 				},
 			},
-			allowed: false,
+			allowed: true,
 		},
 		"ServiceAccountName changed": {
 			orig: &fullSpec,
@@ -164,6 +234,45 @@ func TestKafkaSourceCheckImmutableFields(t *testing.T) {
 			updated: fullSpec,
 			allowed: true,
 		},
+		"consumerGroup changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: "no-way",
+			},
+			allowed: false,
+		},
+		"Kafka TLS Spec changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: fullSpec.ConsumerGroup,
+				SourceSpec:    fullSpec.SourceSpec,
+				Topics:        fullSpec.Topics,
+				KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+					Net: bindingsv1beta1.KafkaNetSpec{
+						TLS: bindingsv1beta1.KafkaTLSSpec{
+							Enable: true,
+						},
+					},
+				},
+			},
+			allowed: true,
+		},
+		"Kafka SASL Spec changed": {
+			orig: &fullSpec,
+			updated: KafkaSourceSpec{
+				ConsumerGroup: fullSpec.ConsumerGroup,
+				SourceSpec:    fullSpec.SourceSpec,
+				Topics:        fullSpec.Topics,
+				KafkaAuthSpec: bindingsv1beta1.KafkaAuthSpec{
+					Net: bindingsv1beta1.KafkaNetSpec{
+						SASL: bindingsv1beta1.KafkaSASLSpec{
+							Enable: true,
+						},
+					},
+				},
+			},
+			allowed: true,
+		},
 	}
 
 	for n, tc := range testCases {
diff --git a/pkg/channel/distributed/common/config/configwatcher_test.go b/pkg/channel/distributed/common/config/configwatcher_test.go
index eff763eeb..fda8d2d86 100644
--- a/pkg/channel/distributed/common/config/configwatcher_test.go
+++ b/pkg/channel/distributed/common/config/configwatcher_test.go
@@ -39,7 +39,7 @@ var (
 	configMapMutex   = sync.Mutex{} // Don't trip up the data race examiner during tests
 )
 
-// Test The InitializeObservability() Functionality
+// Test The InitializeConfigWatcher() Functionality
 func TestInitializeConfigWatcher(t *testing.T) {
 
 	// Obtain a Test Logger (Required By be InitializeConfigWatcher function)
@@ -48,13 +48,13 @@ func TestInitializeConfigWatcher(t *testing.T) {
 	// Setup Environment
 	commontesting.SetTestEnvironment(t)
 
-	// Create A Test Observability ConfigMap For The InitializeObservability() Call To Watch
+	// Create A Test Sarama ConfigMap For The InitializeConfigWatcher() Call To Watch
 	configMap := commontesting.GetTestSaramaConfigMap(commontesting.OldSaramaConfig, commontesting.TestEKConfig)
 
 	// Create The Fake K8S Client And Add It To The ConfigMap
 	fakeK8sClient := fake.NewSimpleClientset(configMap)
 
-	// Add The Fake K8S Client To The Context (Required By InitializeObservability)
+	// Add The Fake K8S Client To The Context (Required By InitializeConfigWatcher)
 	ctx := context.WithValue(context.TODO(), injectionclient.Key{}, fakeK8sClient)
 
 	// The configWatcherHandler should change the nil "watchedConfigMap" to a valid ConfigMap when the watcher triggers
@@ -63,10 +63,14 @@ func TestInitializeConfigWatcher(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, testConfigMap.Data["sarama"], commontesting.OldSaramaConfig)
 
-	// Perform The Test (Initialize The Observability Watcher)
+	// Perform The Test (Initialize The Config Watcher)
 	err = InitializeConfigWatcher(ctx, logger, configWatcherHandler, system.Namespace())
 	assert.Nil(t, err)
 
+	// Note that this initial change to the watched map is not part of the default Kubernetes watching logic.
+	// The underlying KNative InformedWatcher Start() function that is called as part of the
+	// SetupConfigMapWatchOrDie code "pretends" that all watched resources were just created.
+
 	// Wait for the configWatcherHandler to be called (happens pretty quickly; loop usually only runs once)
 	for try := 0; getWatchedMap() == nil && try < 100; try++ {
 		time.Sleep(5 * time.Millisecond)
@@ -105,7 +109,7 @@ func setWatchedMap(configMap *corev1.ConfigMap) {
 }
 
 // Handler function for the ConfigMap watcher
-func configWatcherHandler(ctx context.Context, configMap *corev1.ConfigMap) {
+func configWatcherHandler(_ context.Context, configMap *corev1.ConfigMap) {
 	// Set the package variable to indicate that the test watcher was called
 	setWatchedMap(configMap)
 }
diff --git a/pkg/channel/distributed/common/config/secretwatcher_test.go b/pkg/channel/distributed/common/config/secretwatcher_test.go
new file mode 100644
index 000000000..9794c4e12
--- /dev/null
+++ b/pkg/channel/distributed/common/config/secretwatcher_test.go
@@ -0,0 +1,189 @@
+/*
+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 config
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+	kafkaconstants "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
+	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
+	"knative.dev/eventing-kafka/pkg/common/constants"
+	injectionclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/pkg/system"
+)
+
+// The watcher handler sets this variable to indicate that it was called
+var (
+	watchedSecret *corev1.Secret
+	secretMutex   = sync.Mutex{} // Don't trip up the data race examiner during tests
+)
+
+func TestGetAuthConfigFromKubernetes(t *testing.T) {
+	// Setup Test Environment Namespaces
+	commontesting.SetTestEnvironment(t)
+
+	// Test Data
+	oldAuthSecret := getSaramaTestSecret(t,
+		commontesting.SecretName,
+		commontesting.OldAuthUsername,
+		commontesting.OldAuthPassword,
+		commontesting.OldAuthNamespace,
+		commontesting.OldAuthSaslType)
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		name    string
+		secret  *corev1.Secret
+		askName string
+		wantErr bool
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			name:    "Valid secret, valid request",
+			secret:  oldAuthSecret,
+			askName: commontesting.SecretName,
+			wantErr: false,
+		},
+		{
+			name:    "Valid secret, not found",
+			secret:  oldAuthSecret,
+			askName: "invalid-secret-name",
+			wantErr: true,
+		},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			ctx := context.WithValue(context.TODO(), injectionclient.Key{}, fake.NewSimpleClientset(testCase.secret))
+			kafkaAuth, err := GetAuthConfigFromKubernetes(ctx, testCase.askName, commontesting.SystemNamespace)
+			if !testCase.wantErr {
+				assert.Nil(t, err)
+				assert.NotNil(t, kafkaAuth)
+				assert.Equal(t, commontesting.OldAuthUsername, kafkaAuth.SASL.User)
+				assert.Equal(t, commontesting.OldAuthPassword, kafkaAuth.SASL.Password)
+				assert.Equal(t, commontesting.OldAuthSaslType, kafkaAuth.SASL.SaslType)
+			} else {
+				assert.NotNil(t, err)
+			}
+		})
+	}
+}
+
+func TestGetConfigFromSecret_Valid(t *testing.T) {
+	secret := getSaramaTestSecret(t,
+		commontesting.SecretName,
+		commontesting.OldAuthUsername,
+		commontesting.OldAuthPassword,
+		commontesting.OldAuthNamespace,
+		commontesting.OldAuthSaslType)
+	kafkaAuth := GetAuthConfigFromSecret(secret)
+	assert.Equal(t, commontesting.OldAuthUsername, kafkaAuth.SASL.User)
+	assert.Equal(t, commontesting.OldAuthPassword, kafkaAuth.SASL.Password)
+	assert.Equal(t, commontesting.OldAuthSaslType, kafkaAuth.SASL.SaslType)
+}
+
+func TestGetConfigFromSecret_Invalid(t *testing.T) {
+	authCfg := GetAuthConfigFromSecret(nil)
+	assert.Nil(t, authCfg)
+}
+
+// Test The InitializeSecretWatcher() Functionality
+func TestInitializeSecretWatcher(t *testing.T) {
+	secret := getSaramaTestSecret(t,
+		constants.SettingsSecretName,
+		commontesting.OldAuthUsername,
+		commontesting.OldAuthPassword,
+		commontesting.OldAuthNamespace,
+		commontesting.OldAuthSaslType)
+	fakeK8sClient := fake.NewSimpleClientset(secret)
+	ctx := context.WithValue(context.TODO(), injectionclient.Key{}, fakeK8sClient)
+	ctx, cancel := context.WithCancel(ctx)
+
+	// The secretWatcherHandler should change the nil "watchedSecret" to a valid Secret when the watcher triggers
+
+	testSecret, err := fakeK8sClient.CoreV1().Secrets(system.Namespace()).Get(ctx, constants.SettingsSecretName, metav1.GetOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyUsername]), commontesting.OldAuthUsername)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyPassword]), commontesting.OldAuthPassword)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyNamespace]), commontesting.OldAuthNamespace)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeySaslType]), commontesting.OldAuthSaslType)
+
+	// Perform The Test (Initialize The Secret Watcher)
+	err = InitializeSecretWatcher(ctx, system.Namespace(), constants.SettingsSecretName, secretWatcherHandler)
+	assert.Nil(t, err)
+
+	// The secretWatcherHandler should change this back to a valid Secret after the watcher is triggered
+	setWatchedSecret(nil)
+
+	// Change the data in the secret
+	testSecret.Data[kafkaconstants.KafkaSecretKeyUsername] = []byte(commontesting.NewAuthUsername)
+	testSecret.Data[kafkaconstants.KafkaSecretKeyPassword] = []byte(commontesting.NewAuthPassword)
+	testSecret.Data[kafkaconstants.KafkaSecretKeyNamespace] = []byte(commontesting.NewAuthNamespace)
+	testSecret.Data[kafkaconstants.KafkaSecretKeySaslType] = []byte(commontesting.NewAuthSaslType)
+
+	// Update the secret in the Kubernetes client
+	testSecret, err = fakeK8sClient.CoreV1().Secrets(system.Namespace()).Update(ctx, testSecret, metav1.UpdateOptions{})
+	assert.Nil(t, err)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyPassword]), commontesting.NewAuthPassword)
+
+	// Wait for the secretWatcherHandler to be called (happens pretty quickly; loop usually only runs once)
+	for try := 0; getWatchedSecret() == nil && try < 100; try++ {
+		time.Sleep(5 * time.Millisecond)
+	}
+	assert.NotNil(t, getWatchedSecret())
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyUsername]), commontesting.NewAuthUsername)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyPassword]), commontesting.NewAuthPassword)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeyNamespace]), commontesting.NewAuthNamespace)
+	assert.Equal(t, string(testSecret.Data[kafkaconstants.KafkaSecretKeySaslType]), commontesting.NewAuthSaslType)
+
+	// End the watcher
+	cancel()
+}
+
+func getWatchedSecret() *corev1.Secret {
+	secretMutex.Lock()
+	defer secretMutex.Unlock()
+	return watchedSecret
+}
+
+func setWatchedSecret(secret *corev1.Secret) {
+	secretMutex.Lock()
+	watchedSecret = secret
+	defer secretMutex.Unlock()
+}
+
+// Handler function for the Secret watcher
+func secretWatcherHandler(_ context.Context, secret *corev1.Secret) {
+	// Set the package variable to indicate that the test watcher was called
+	setWatchedSecret(secret)
+}
+
+func getSaramaTestSecret(t *testing.T, name string,
+	username string, password string, namespace string, saslType string) *corev1.Secret {
+	commontesting.SetTestEnvironment(t)
+	return commontesting.GetTestSaramaSecret(name, username, password, namespace, saslType)
+}
diff --git a/pkg/channel/distributed/common/config/testing/util.go b/pkg/channel/distributed/common/config/testing/util.go
index 0698b100b..fddca255a 100644
--- a/pkg/channel/distributed/common/config/testing/util.go
+++ b/pkg/channel/distributed/common/config/testing/util.go
@@ -21,6 +21,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kafkaconstants "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/system"
 )
@@ -66,12 +67,18 @@ dispatcher:
   replicas: 1
 kafka:
   enableSaramaLogging: false
+  brokers: ` + DefaultKafkaBroker + `
   topic:
     defaultNumPartitions: 4
     defaultReplicationFactor: 1
     defaultRetentionMillis: 604800000
     adminType: kafka
 `
+	DefaultKafkaBroker     = "TestBroker"
+	DefaultSecretUsername  = "TestUsername"
+	DefaultSecretPassword  = "TestPassword"
+	DefaultSecretSaslType  = "PLAIN"
+	DefaultSecretNamespace = "TestNamespace"
 )
 
 // KafkaConfigMapOption Enables Customization Of An Eventing-Kafka ConfigMap
@@ -149,3 +156,66 @@ func WithModifiedSaramaProducer(configMap *corev1.ConfigMap) {
 func WithoutEventingKafkaConfiguration(configMap *corev1.ConfigMap) {
 	delete(configMap.Data, constants.EventingKafkaSettingsConfigKey)
 }
+
+// KafkaConfigMapOption Enables Customization Of An Eventing-Kafka Secret
+type KafkaSecretOption func(secret *corev1.Secret)
+
+// Create A New Eventing-Kafka Secret For Testing
+func NewKafkaSecret(options ...KafkaSecretOption) *corev1.Secret {
+
+	// Create A Base Kafka Secret With Default Auth Configuration
+	secret := &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Secret",
+			APIVersion: corev1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      constants.SettingsSecretName,
+			Namespace: system.Namespace(),
+		},
+		Data: map[string][]byte{
+			kafkaconstants.KafkaSecretKeyPassword:  []byte(DefaultSecretPassword),
+			kafkaconstants.KafkaSecretKeySaslType:  []byte(DefaultSecretSaslType),
+			kafkaconstants.KafkaSecretKeyUsername:  []byte(DefaultSecretUsername),
+			kafkaconstants.KafkaSecretKeyNamespace: []byte(DefaultSecretNamespace),
+		},
+	}
+
+	// Apply The Specified Eventing-Kafka ConfigMap Options
+	for _, option := range options {
+		option(secret)
+	}
+
+	// Return The Custom Eventing-Kafka Secret
+	return secret
+}
+
+// Modify The Default Password Section Of The Secret Data
+func WithModifiedPassword(secret *corev1.Secret) {
+	secret.Data[kafkaconstants.KafkaSecretKeyPassword] = []byte("TestModifiedPassword")
+}
+
+// Modify The Default Username Section Of The Secret Data
+func WithModifiedUsername(secret *corev1.Secret) {
+	secret.Data[kafkaconstants.KafkaSecretKeyUsername] = []byte("TestModifiedUsername")
+}
+
+// Empty The Default Username Section Of The Secret Data
+func WithEmptyUsername(secret *corev1.Secret) {
+	secret.Data[kafkaconstants.KafkaSecretKeyUsername] = []byte("")
+}
+
+// Modify The Default SaslType Section Of The Secret Data
+func WithModifiedSaslType(secret *corev1.Secret) {
+	secret.Data[kafkaconstants.KafkaSecretKeySaslType] = []byte("TestModifiedSaslType")
+}
+
+// Modify The Default Namespace Section Of The Secret Data
+func WithModifiedNamespace(secret *corev1.Secret) {
+	secret.Data[kafkaconstants.KafkaSecretKeyNamespace] = []byte("TestModifiedNamespace")
+}
+
+// Remove the Data From The Secret
+func WithMissingConfig(secret *corev1.Secret) {
+	secret.Data = nil
+}
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index 9541e9fd5..f8dd508e7 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -199,6 +199,59 @@ func TestLoadSettings(t *testing.T) {
 	assert.NotNil(t, err)
 }
 
+func TestAuthFromSarama(t *testing.T) {
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		name             string
+		existingUser     string
+		existingPassword string
+		existingSaslType sarama.SASLMechanism
+		expectNil        bool
+		expectedUser     string
+		expectedPassword string
+		expectedSaslType string
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			name:             "Empty User",
+			existingUser:     "",
+			existingPassword: "testPassword",
+			existingSaslType: sarama.SASLTypeOAuth,
+			expectNil:        true,
+		},
+		{
+			name:             "Existing User",
+			existingUser:     "testUser",
+			existingPassword: "testPassword",
+			existingSaslType: sarama.SASLTypePlaintext,
+			expectedUser:     "testUser",
+			expectedPassword: "testPassword",
+			expectedSaslType: sarama.SASLTypePlaintext,
+		},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			existingConfig := sarama.NewConfig()
+			existingConfig.Net.SASL.User = testCase.existingUser
+			existingConfig.Net.SASL.Password = testCase.existingPassword
+			existingConfig.Net.SASL.Mechanism = testCase.existingSaslType
+			config := AuthFromSarama(existingConfig)
+			if testCase.expectNil {
+				assert.Nil(t, config)
+			} else {
+				assert.Equal(t, testCase.expectedUser, config.SASL.User)
+				assert.Equal(t, testCase.expectedPassword, config.SASL.Password)
+				assert.Equal(t, testCase.expectedSaslType, config.SASL.SaslType)
+			}
+		})
+	}
+}
+
 func verifyTestEKConfigSettings(t *testing.T, saramaConfig *sarama.Config, eventingKafkaConfig *commonconfig.EventingKafkaConfig) {
 	// Quick checks to make sure the loaded configs aren't complete junk
 	assert.Equal(t, commontesting.OldUsername, saramaConfig.Net.SASL.User)
diff --git a/pkg/channel/distributed/common/testing/constants.go b/pkg/channel/distributed/common/testing/constants.go
index 9eae68f63..0fed8fcaa 100644
--- a/pkg/channel/distributed/common/testing/constants.go
+++ b/pkg/channel/distributed/common/testing/constants.go
@@ -18,8 +18,8 @@ package testing
 
 const (
 	SystemNamespace = "eventing-test-ns"
+	SecretName      = "test-secret-name"
 
-	OldClientId = "TestOldClientId"
 	NewClientId = "TestNewClientId"
 	OldUsername = "TestOldUsername"
 	NewUsername = "TestNewUsername"
@@ -30,16 +30,6 @@ const (
 	DispatcherRetryInitial = "5000"
 	DispatcherRetry        = "500000"
 
-	// These constants are used here to make sure that the CreateConsumerGroup() call doesn't have problems,
-	// but they aren't non-testing defaults since most settings are now in 200-eventing-kafka-configmap.yaml
-	ConfigAdminTimeout                      = "10000000000"
-	ConfigNetKeepAlive                      = "30000000000"
-	ConfigMetadataRefreshFrequency          = "300000000000"
-	ConfigConsumerOffsetsAutoCommitInterval = "5000000000"
-	ConfigConsumerOffsetsRetention          = "604800000000000"
-	ConfigProducerIdempotent                = "false"
-	ConfigProducerRequiredAcks              = "-1"
-
 	TestEKConfig = `
 dispatcher:
   cpuLimit: 500m
@@ -82,24 +72,13 @@ Metadata:
 ClientID: ` + NewClientId + `
 `
 
-	SaramaDefaultConfigYaml = `
-Admin:
-  Timeout: ` + ConfigAdminTimeout + `
-Net:
-  KeepAlive: ` + ConfigNetKeepAlive + `
-Metadata:
-  RefreshFrequency: ` + ConfigMetadataRefreshFrequency + `
-Consumer:
-  Offsets:
-    AutoCommit:
-      Interval: ` + ConfigConsumerOffsetsAutoCommitInterval + `
-    Retention: ` + ConfigConsumerOffsetsRetention + `
-  Return:
-    Errors: true
-Producer:
-  Idempotent: ` + ConfigProducerIdempotent + `
-  RequiredAcks: ` + ConfigProducerRequiredAcks + `
-  Return:
-    Successes: true
-`
+	OldAuthNamespace = "TestOldNamespace"
+	OldAuthPassword  = "TestOldPassword"
+	OldAuthSaslType  = "TestOldAuthSaslType"
+	OldAuthUsername  = "TestOldAuthUsername"
+
+	NewAuthNamespace = "TestNewNamespace"
+	NewAuthPassword  = "TestNewPassword"
+	NewAuthSaslType  = "TestNewAuthSaslType"
+	NewAuthUsername  = "TestNewAuthUsername"
 )
diff --git a/pkg/channel/distributed/common/testing/util.go b/pkg/channel/distributed/common/testing/util.go
index 6a5fb0b69..48b18bf45 100644
--- a/pkg/channel/distributed/common/testing/util.go
+++ b/pkg/channel/distributed/common/testing/util.go
@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/env"
+	kafkaconstants "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants"
 	"knative.dev/eventing-kafka/pkg/common/constants"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/system"
@@ -34,6 +35,27 @@ func GetTestSaramaConfigMap(saramaConfig string, configuration string) *corev1.C
 	return GetTestSaramaConfigMapNamespaced(constants.SettingsConfigMapName, system.Namespace(), saramaConfig, configuration)
 }
 
+// Returns A Secret Containing The Desired Fields
+func GetTestSaramaSecret(name string, username string, password string, namespace string, saslType string) *corev1.Secret {
+	return &corev1.Secret{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "Secret",
+			APIVersion: corev1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: system.Namespace(),
+			Labels:    map[string]string{kafkaconstants.KafkaSecretLabel: "true"},
+		},
+		Data: map[string][]byte{
+			kafkaconstants.KafkaSecretKeyUsername:  []byte(username),
+			kafkaconstants.KafkaSecretKeyPassword:  []byte(password),
+			kafkaconstants.KafkaSecretKeyNamespace: []byte(namespace),
+			kafkaconstants.KafkaSecretKeySaslType:  []byte(saslType),
+		},
+	}
+}
+
 // Returns A ConfigMap Containing The Desired Sarama Config YAML, Name And Namespace
 func GetTestSaramaConfigMapNamespaced(name, namespace, saramaConfig, configuration string) *corev1.ConfigMap {
 	return &corev1.ConfigMap{
diff --git a/pkg/channel/distributed/controller/testing/data.go b/pkg/channel/distributed/controller/testing/data.go
index 42f1eb86e..20b7f95b0 100644
--- a/pkg/channel/distributed/controller/testing/data.go
+++ b/pkg/channel/distributed/controller/testing/data.go
@@ -287,10 +287,9 @@ func NewKafkaSecret(options ...KafkaSecretOption) *corev1.Secret {
 			Labels:    map[string]string{kafkaconstants.KafkaSecretLabel: "true"},
 		},
 		Data: map[string][]byte{
-			constants.KafkaSecretDataKeyBrokers:  []byte(KafkaSecretDataValueBrokers),
-			constants.KafkaSecretDataKeyUsername: []byte(KafkaSecretDataValueUsername),
-			constants.KafkaSecretDataKeyPassword: []byte(KafkaSecretDataValuePassword),
-			constants.KafkaSecretDataKeySaslType: []byte(KafkaSecretDataValueSaslType),
+			kafkaconstants.KafkaSecretKeyUsername: []byte(KafkaSecretDataValueUsername),
+			kafkaconstants.KafkaSecretKeyPassword: []byte(KafkaSecretDataValuePassword),
+			kafkaconstants.KafkaSecretKeySaslType: []byte(KafkaSecretDataValueSaslType),
 		},
 		Type: "opaque",
 	}
@@ -471,7 +470,7 @@ func WithReceiverDeploymentFinalized(kafkachannel *kafkav1beta1.KafkaChannel) {
 }
 
 // Set The KafkaChannel's Dispatcher Deployment As READY
-func WithDispatcherDeploymentReady(kafkachannel *kafkav1beta1.KafkaChannel) {
+func WithDispatcherDeploymentReady(_ *kafkav1beta1.KafkaChannel) {
 	// TODO - This is unnecessary since the testing framework doesn't return any Status Conditions from the K8S commands (Create, Get)
 	//        which means the propagate function doesn't do anything.  This is a testing gap with the framework and propagateDispatcherStatus()
 	// kafkachannel.Status.PropagateDispatcherStatus()
@@ -579,9 +578,6 @@ 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{
@@ -686,41 +682,12 @@ func NewKafkaChannelReceiverDeployment(options ...DeploymentOption) *appsv1.Depl
 									Value: strconv.Itoa(int(ResyncPeriod / time.Minute)),
 								},
 								{
-									Name: commonenv.KafkaBrokerEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyBrokers,
-										},
-									},
-								},
-								{
-									Name: commonenv.KafkaUsernameEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyUsername,
-										},
-									},
+									Name:  commonenv.KafkaSecretNameEnvVarKey,
+									Value: KafkaSecretName,
 								},
 								{
-									Name: commonenv.KafkaPasswordEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyPassword,
-										},
-									},
-								},
-								{
-									Name: commonenv.KafkaSaslTypeEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeySaslType,
-											Optional:             &optional,
-										},
-									},
+									Name:  commonenv.KafkaSecretNamespaceEnvVarKey,
+									Value: KafkaSecretNamespace,
 								},
 							},
 							ImagePullPolicy: corev1.PullIfNotPresent,
@@ -806,7 +773,6 @@ 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)
@@ -916,41 +882,12 @@ func NewKafkaChannelDispatcherDeployment(options ...DeploymentOption) *appsv1.De
 									Value: strconv.Itoa(int(ResyncPeriod / time.Minute)),
 								},
 								{
-									Name: commonenv.KafkaBrokerEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyBrokers,
-										},
-									},
-								},
-								{
-									Name: commonenv.KafkaUsernameEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyUsername,
-										},
-									},
+									Name:  commonenv.KafkaSecretNameEnvVarKey,
+									Value: KafkaSecretName,
 								},
 								{
-									Name: commonenv.KafkaPasswordEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeyPassword,
-										},
-									},
-								},
-								{
-									Name: commonenv.KafkaSaslTypeEnvVarKey,
-									ValueFrom: &corev1.EnvVarSource{
-										SecretKeyRef: &corev1.SecretKeySelector{
-											LocalObjectReference: corev1.LocalObjectReference{Name: KafkaSecretName},
-											Key:                  constants.KafkaSecretDataKeySaslType,
-											Optional:             &optional,
-										},
-									},
+									Name:  commonenv.KafkaSecretNamespaceEnvVarKey,
+									Value: KafkaSecretNamespace,
 								},
 							},
 							ImagePullPolicy: corev1.PullIfNotPresent,
diff --git a/pkg/channel/distributed/controller/util/secret_test.go b/pkg/channel/distributed/controller/util/secret_test.go
index baeb4c360..b59fcc445 100644
--- a/pkg/channel/distributed/controller/util/secret_test.go
+++ b/pkg/channel/distributed/controller/util/secret_test.go
@@ -103,69 +103,6 @@ func TestGetKafkaSecret(t *testing.T) {
 	assert.Nil(t, result)
 }
 
-// Test The ValidateKafkaSecret() Functionality
-func TestValidateKafkaSecret(t *testing.T) {
-
-	// Test Data
-	brokers := "TestBrokers"
-	username := "TestUsername"
-	password := "TestPassword"
-
-	// Create A Test Logger
-	logger := logtesting.TestLogger(t).Desugar()
-
-	// Define & Create The Test Cases
-	tests := []struct {
-		name string
-		data map[string][]byte
-		want bool
-	}{
-		{
-			name: "Valid Kafka Secret (No Auth)",
-			data: map[string][]byte{
-				clientconstants.KafkaSecretKeyBrokers:  []byte(brokers),
-				clientconstants.KafkaSecretKeyUsername: []byte(""),
-				clientconstants.KafkaSecretKeyPassword: []byte(""),
-			},
-			want: true,
-		},
-		{
-			name: "Valid Kafka Secret (Auth)",
-			data: map[string][]byte{
-				clientconstants.KafkaSecretKeyBrokers:  []byte(brokers),
-				clientconstants.KafkaSecretKeyUsername: []byte(username),
-				clientconstants.KafkaSecretKeyPassword: []byte(password),
-			},
-			want: true,
-		},
-		{
-			name: "Invalid Kafka Secret (No Brokers)",
-			data: map[string][]byte{
-				clientconstants.KafkaSecretKeyUsername: []byte(username),
-				clientconstants.KafkaSecretKeyPassword: []byte(password),
-			},
-			want: false,
-		},
-	}
-
-	// Loop Over The Test Cases
-	for _, tt := range tests {
-
-		// Run Each Test Case
-		t.Run(tt.name, func(t *testing.T) {
-
-			// Kafka Secret To Test
-			secret := &corev1.Secret{Data: tt.data}
-
-			// Perform The Test
-			result := ValidateKafkaSecret(logger, secret)
-
-			// Verify The Results
-			assert.Equal(t, tt.want, result)
-		})
-	}
-}
-
 // Create K8S Kafka Secret With Specified Config
 func createKafkaSecret(name string, namespace string) *corev1.Secret {
 	return &corev1.Secret{
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 3fb6d5b4f..175285221 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -187,7 +187,8 @@ func populateEnvironmentVariables(t *testing.T) {
 	assert.Nil(t, os.Setenv(commonenv.PodNameEnvVarKey, "testPodName"))
 	assert.Nil(t, os.Setenv(commonenv.ContainerNameEnvVarKey, "testContainerName"))
 	assert.Nil(t, os.Setenv(commonenv.HealthPortEnvVarKey, "5678"))
-	assert.Nil(t, os.Setenv(commonenv.KafkaBrokerEnvVarKey, "testKafkaBroker"))
+	assert.Nil(t, os.Setenv(commonenv.KafkaSecretNameEnvVarKey, "testKafkaSecretName"))
+	assert.Nil(t, os.Setenv(commonenv.KafkaSecretNamespaceEnvVarKey, "testKafkaSecretNamespace"))
 	assert.Nil(t, os.Setenv(commonenv.KafkaTopicEnvVarKey, "testKafkaTopic"))
 	assert.Nil(t, os.Setenv(commonenv.ChannelKeyEnvVarKey, "testChannelKey"))
 	assert.Nil(t, os.Setenv(commonenv.ServiceNameEnvVarKey, "testServiceName"))
@@ -217,6 +218,10 @@ func (m MockDispatcher) UpdateSubscriptions(_ []eventingduck.SubscriberSpec) map
 	return nil
 }
 
-func (m MockDispatcher) ConfigChanged(context.Context, *corev1.ConfigMap) dispatcher.Dispatcher {
+func (m MockDispatcher) ConfigChanged(_ context.Context, _ *corev1.ConfigMap) dispatcher.Dispatcher {
+	return nil
+}
+
+func (m MockDispatcher) SecretChanged(_ context.Context, _ *corev1.Secret) dispatcher.Dispatcher {
 	return nil
 }
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index a1b260773..4fe9dd191 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -32,8 +32,6 @@ import (
 	consumerwrapper "knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/consumer/wrapper"
 	"knative.dev/eventing-kafka/pkg/channel/distributed/common/metrics"
 	commontesting "knative.dev/eventing-kafka/pkg/channel/distributed/common/testing"
-	dispatchertesting "knative.dev/eventing-kafka/pkg/channel/distributed/dispatcher/testing"
-	receivertesting "knative.dev/eventing-kafka/pkg/channel/distributed/receiver/testing"
 	commonclient "knative.dev/eventing-kafka/pkg/common/client"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/logging"
@@ -124,7 +122,7 @@ func TestUpdateSubscriptions(t *testing.T) {
 	defer consumertesting.RestoreNewConsumerGroupFn()
 
 	// Test Data
-	brokers := []string{dispatchertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	config, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build(ctx)
 	assert.Nil(t, err)
 
@@ -292,7 +290,7 @@ func TestConfigChanged(t *testing.T) {
 	commontesting.SetTestEnvironment(t)
 
 	// Test Data
-	brokers := []string{receivertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	baseSaramaConfig, err := commonclient.NewConfigBuilder().
 		WithDefaults().
 		FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).
@@ -381,6 +379,113 @@ func TestConfigChanged(t *testing.T) {
 	}
 }
 
+// Test The Dispatcher's SecretChanged Functionality
+func TestSecretChanged(t *testing.T) {
+
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
+	// Setup Test Environment Namespaces
+	commontesting.SetTestEnvironment(t)
+
+	// Test Data
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
+	auth := &commonclient.KafkaAuthConfig{
+		SASL: &commonclient.KafkaSaslConfig{
+			User:     commonconfigtesting.DefaultSecretUsername,
+			Password: commonconfigtesting.DefaultSecretPassword,
+			SaslType: commonconfigtesting.DefaultSecretSaslType,
+		},
+	}
+	baseSaramaConfig, err := commonclient.NewConfigBuilder().
+		WithDefaults().
+		FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).
+		WithVersion(&sarama.V2_0_0_0).
+		WithAuth(auth).
+		Build(ctx)
+	assert.Nil(t, err)
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		only                bool
+		name                string
+		newSecret           *corev1.Secret
+		expectNewDispatcher bool
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			name:                "No Changes (Same Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(),
+			expectNewDispatcher: false,
+		},
+		{
+			name:                "Password Change (New Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedPassword),
+			expectNewDispatcher: true,
+		},
+		{
+			name:                "Username Change (New Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedUsername),
+			expectNewDispatcher: true,
+		},
+		{
+			name:                "Empty Username Change (New Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithEmptyUsername),
+			expectNewDispatcher: true,
+		},
+		{
+			name:                "SaslType Change (New Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedSaslType),
+			expectNewDispatcher: true,
+		},
+		{
+			name:                "Namespace Change (Same Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedNamespace),
+			expectNewDispatcher: false,
+		},
+		{
+			name:                "No Auth Config In Secret (Same Dispatcher)",
+			newSecret:           configtesting.NewKafkaSecret(configtesting.WithMissingConfig),
+			expectNewDispatcher: false,
+		},
+	}
+
+	// Filter To Those With "only" Flag (If Any Specified)
+	filteredTestCases := make([]TestCase, 0)
+	for _, testCase := range testCases {
+		if testCase.only {
+			filteredTestCases = append(filteredTestCases, testCase)
+		}
+	}
+	if len(filteredTestCases) == 0 {
+		filteredTestCases = testCases
+	}
+
+	// Make Sure To Restore The NewConsumerGroup Wrapper After The Test
+	defer consumertesting.RestoreNewConsumerGroupFn()
+
+	// Run The Filtered TestCases
+	for _, testCase := range filteredTestCases {
+		t.Run(testCase.name, func(t *testing.T) {
+
+			// Mock The SyncProducer & Stub The NewConsumerGroupWrapper()
+			mockConsumerGroup := consumertesting.NewMockConsumerGroup()
+			consumertesting.StubNewConsumerGroupFn(consumertesting.NonValidatingNewConsumerGroupFn(mockConsumerGroup))
+
+			// Create A Test Dispatcher To Perform Tests Against
+			dispatcher := createTestDispatcher(t, brokers, baseSaramaConfig)
+
+			// Perform The Test
+			newDispatcher := dispatcher.SecretChanged(ctx, testCase.newSecret)
+
+			// Verify Expected State (Not Much To Verify Due To Interface)
+			assert.Equal(t, testCase.expectNewDispatcher, newDispatcher != nil)
+		})
+	}
+}
+
 // Utility Function For Creating A SubscriberWrapper With Specified UID & Mock ConsumerGroup
 func createSubscriberWrapper(uid types.UID) *SubscriberWrapper {
 	return NewSubscriberWrapper(eventingduck.SubscriberSpec{UID: uid}, fmt.Sprintf("kafka.%s", string(uid)), nil)
diff --git a/pkg/channel/distributed/dispatcher/env/environment_test.go b/pkg/channel/distributed/dispatcher/env/environment_test.go
index eead5cdef..3800b4791 100644
--- a/pkg/channel/distributed/dispatcher/env/environment_test.go
+++ b/pkg/channel/distributed/dispatcher/env/environment_test.go
@@ -33,20 +33,18 @@ import (
 
 // Test Constants
 const (
-	systemNamespace = "test-system-namespace"
-	metricsPort     = "9999"
-	metricsDomain   = "kafka-eventing"
-	healthPort      = "1234"
-	resyncPeriod    = "3600"
-	kafkaBrokers    = "TestKafkaBrokers"
-	kafkaTopic      = "TestKafkaTopic"
-	channelKey      = "TestChannelKey"
-	serviceName     = "TestServiceName"
-	kafkaUsername   = "TestKafkaUsername"
-	kafkaPassword   = "TestKafkaPassword"
-	kafkaSaslType   = "SCRAM-SHA-512"
-	podName         = "TestPod"
-	containerName   = "TestContainer"
+	systemNamespace      = "test-system-namespace"
+	metricsPort          = "9999"
+	metricsDomain        = "kafka-eventing"
+	healthPort           = "1234"
+	resyncPeriod         = "3600"
+	kafkaTopic           = "TestKafkaTopic"
+	channelKey           = "TestChannelKey"
+	serviceName          = "TestServiceName"
+	kafkaSecretName      = "TestKafkaUsername"
+	kafkaSecretNamespace = "TestKafkaPassword"
+	podName              = "TestPod"
+	containerName        = "TestContainer"
 )
 
 // Define The TestCase Struct
@@ -57,13 +55,11 @@ type TestCase struct {
 	metricsDomain        string
 	healthPort           string
 	resyncPeriodMinutes  string
-	kafkaBrokers         string
 	kafkaTopic           string
 	channelKey           string
 	serviceName          string
-	kafkaUsername        string
-	kafkaPassword        string
-	kafkaSaslType        string
+	kafkaSecretName      string
+	kafkaSecretNamespace string
 	podName              string
 	containerName        string
 	expectedError        error
@@ -111,9 +107,14 @@ func TestGetEnvironment(t *testing.T) {
 	testCase.expectedError = getInvalidIntEnvironmentVariableError(testCase.healthPort, commonenv.HealthPortEnvVarKey)
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Missing Required Config - KafkaBrokers")
-	testCase.kafkaBrokers = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.KafkaBrokerEnvVarKey)
+	testCase = getValidTestCase("Missing Required Config - KafkaSecretName")
+	testCase.kafkaSecretName = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.KafkaSecretNameEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Missing Required Config - KafkaSecretNamespace")
+	testCase.kafkaSecretNamespace = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(commonenv.KafkaSecretNamespaceEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Missing Required Config - KafkaTopic")
@@ -162,13 +163,11 @@ func TestGetEnvironment(t *testing.T) {
 			assertSetenvNonempty(t, commonenv.MetricsPortEnvVarKey, testCase.metricsPort)
 			assertSetenvNonempty(t, commonenv.HealthPortEnvVarKey, testCase.healthPort)
 			assertSetenvNonempty(t, commonenv.ResyncPeriodMinutesEnvVarKey, testCase.resyncPeriodMinutes)
-			assertSetenv(t, commonenv.KafkaBrokerEnvVarKey, testCase.kafkaBrokers)
 			assertSetenv(t, commonenv.KafkaTopicEnvVarKey, testCase.kafkaTopic)
 			assertSetenv(t, commonenv.ChannelKeyEnvVarKey, testCase.channelKey)
 			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.KafkaSecretNameEnvVarKey, testCase.kafkaSecretName)
+			assertSetenv(t, commonenv.KafkaSecretNamespaceEnvVarKey, testCase.kafkaSecretNamespace)
 			assertSetenv(t, commonenv.PodNameEnvVarKey, testCase.podName)
 			assertSetenv(t, commonenv.ContainerNameEnvVarKey, testCase.containerName)
 
@@ -183,13 +182,11 @@ func TestGetEnvironment(t *testing.T) {
 				assert.Equal(t, testCase.systemNamespace, environment.SystemNamespace)
 				assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 				assert.Equal(t, testCase.healthPort, strconv.Itoa(environment.HealthPort))
-				assert.Equal(t, testCase.kafkaBrokers, environment.KafkaBrokers)
 				assert.Equal(t, testCase.kafkaTopic, environment.KafkaTopic)
 				assert.Equal(t, testCase.channelKey, environment.ChannelKey)
 				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.kafkaSecretName, environment.KafkaSecretName)
+				assert.Equal(t, testCase.kafkaSecretNamespace, environment.KafkaSecretNamespace)
 				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)))
@@ -221,13 +218,11 @@ func getValidTestCase(name string) TestCase {
 		metricsDomain:        metricsDomain,
 		healthPort:           healthPort,
 		resyncPeriodMinutes:  resyncPeriod,
-		kafkaBrokers:         kafkaBrokers,
 		kafkaTopic:           kafkaTopic,
 		channelKey:           channelKey,
 		serviceName:          serviceName,
-		kafkaUsername:        kafkaUsername,
-		kafkaPassword:        kafkaPassword,
-		kafkaSaslType:        kafkaSaslType,
+		kafkaSecretName:      kafkaSecretName,
+		kafkaSecretNamespace: kafkaSecretNamespace,
 		podName:              podName,
 		containerName:        containerName,
 		expectedError:        nil,
diff --git a/pkg/channel/distributed/dispatcher/testing/constants.go b/pkg/channel/distributed/dispatcher/testing/constants.go
deleted file mode 100644
index 08e152c5b..000000000
--- a/pkg/channel/distributed/dispatcher/testing/constants.go
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
-Copyright 2020 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 testing
-
-// Test Data
-const (
-	KafkaBroker = "TestBroker"
-)
diff --git a/pkg/channel/distributed/receiver/env/environment_test.go b/pkg/channel/distributed/receiver/env/environment_test.go
index e09bc67f9..a75b3e551 100644
--- a/pkg/channel/distributed/receiver/env/environment_test.go
+++ b/pkg/channel/distributed/receiver/env/environment_test.go
@@ -33,17 +33,16 @@ import (
 
 // Test Constants
 const (
-	systemNamespace     = "test-system-namespace"
-	metricsPort         = "9999"
-	metricsDomain       = "kafka-eventing"
-	healthPort          = "1234"
-	resyncPeriodMinutes = "3600"
-	kafkaBrokers        = "TestKafkaBrokers"
-	serviceName         = "TestServiceName"
-	kafkaUsername       = "TestKafkaUsername"
-	kafkaPassword       = "TestKafkaPassword"
-	podName             = "TestPod"
-	containerName       = "TestContainer"
+	systemNamespace      = "test-system-namespace"
+	metricsPort          = "9999"
+	metricsDomain        = "kafka-eventing"
+	healthPort           = "1234"
+	resyncPeriodMinutes  = "3600"
+	serviceName          = "TestServiceName"
+	kafkaSecretName      = "TestKafkaSecretName"
+	kafkaSecretNamespace = "TestKafkaSecretNamespace"
+	podName              = "TestPod"
+	containerName        = "TestContainer"
 )
 
 // Define The TestCase Struct
@@ -54,10 +53,9 @@ type TestCase struct {
 	metricsDomain        string
 	healthPort           string
 	resyncPeriodMinutes  string
-	kafkaBrokers         string
 	serviceName          string
-	kafkaUsername        string
-	kafkaPassword        string
+	kafkaSecretName      string
+	kafkaSecretNamespace string
 	podName              string
 	containerName        string
 	expectedError        error
@@ -115,9 +113,14 @@ func TestGetEnvironment(t *testing.T) {
 	testCase.expectedResyncPeriod = "600" // 10 hours - default value
 	testCases = append(testCases, testCase)
 
-	testCase = getValidTestCase("Missing Required Config - KafkaBrokers")
-	testCase.kafkaBrokers = ""
-	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.KafkaBrokerEnvVarKey)
+	testCase = getValidTestCase("Missing Required Config - KafkaSecretName")
+	testCase.kafkaSecretName = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.KafkaSecretNameEnvVarKey)
+	testCases = append(testCases, testCase)
+
+	testCase = getValidTestCase("Missing Required Config - KafkaSecretNamespace")
+	testCase.kafkaSecretNamespace = ""
+	testCase.expectedError = getMissingRequiredEnvironmentVariableError(env.KafkaSecretNamespaceEnvVarKey)
 	testCases = append(testCases, testCase)
 
 	testCase = getValidTestCase("Missing Required Config - ServiceName")
@@ -145,10 +148,9 @@ func TestGetEnvironment(t *testing.T) {
 			assertSetenv(t, env.MetricsDomainEnvVarKey, testCase.metricsDomain)
 			assertSetenvNonempty(t, env.MetricsPortEnvVarKey, testCase.metricsPort)
 			assertSetenvNonempty(t, env.HealthPortEnvVarKey, testCase.healthPort)
-			assertSetenv(t, env.KafkaBrokerEnvVarKey, testCase.kafkaBrokers)
+			assertSetenv(t, env.KafkaSecretNameEnvVarKey, testCase.kafkaSecretName)
+			assertSetenv(t, env.KafkaSecretNamespaceEnvVarKey, testCase.kafkaSecretNamespace)
 			assertSetenv(t, env.ServiceNameEnvVarKey, testCase.serviceName)
-			assertSetenv(t, env.KafkaUsernameEnvVarKey, testCase.kafkaUsername)
-			assertSetenv(t, env.KafkaPasswordEnvVarKey, testCase.kafkaPassword)
 			assertSetenv(t, env.PodNameEnvVarKey, testCase.podName)
 			assertSetenv(t, env.ContainerNameEnvVarKey, testCase.containerName)
 			assertSetenvNonempty(t, env.ResyncPeriodMinutesEnvVarKey, testCase.resyncPeriodMinutes)
@@ -164,10 +166,9 @@ func TestGetEnvironment(t *testing.T) {
 				assert.Equal(t, testCase.systemNamespace, environment.SystemNamespace)
 				assert.Equal(t, testCase.metricsPort, strconv.Itoa(environment.MetricsPort))
 				assert.Equal(t, testCase.healthPort, strconv.Itoa(environment.HealthPort))
-				assert.Equal(t, testCase.kafkaBrokers, environment.KafkaBrokers)
 				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.kafkaSecretName, environment.KafkaSecretName)
+				assert.Equal(t, testCase.kafkaSecretNamespace, environment.KafkaSecretNamespace)
 				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)))
@@ -199,10 +200,9 @@ func getValidTestCase(name string) TestCase {
 		metricsDomain:        metricsDomain,
 		healthPort:           healthPort,
 		resyncPeriodMinutes:  resyncPeriodMinutes,
-		kafkaBrokers:         kafkaBrokers,
 		serviceName:          serviceName,
-		kafkaUsername:        kafkaUsername,
-		kafkaPassword:        kafkaPassword,
+		kafkaSecretName:      kafkaSecretName,
+		kafkaSecretNamespace: kafkaSecretNamespace,
 		podName:              podName,
 		containerName:        containerName,
 		expectedResyncPeriod: resyncPeriodMinutes,
diff --git a/pkg/channel/distributed/receiver/producer/producer_test.go b/pkg/channel/distributed/receiver/producer/producer_test.go
index 23c979593..ab077565a 100644
--- a/pkg/channel/distributed/receiver/producer/producer_test.go
+++ b/pkg/channel/distributed/receiver/producer/producer_test.go
@@ -41,7 +41,7 @@ import (
 func TestNewProducer(t *testing.T) {
 
 	// Test Data
-	brokers := []string{receivertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	config := sarama.NewConfig()
 
 	// Create A Mock Kafka SyncProducer
@@ -62,7 +62,7 @@ func TestNewProducer(t *testing.T) {
 func TestProduceKafkaMessage(t *testing.T) {
 
 	// Test Data
-	brokers := []string{receivertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	config := sarama.NewConfig()
 	channelReference := receivertesting.CreateChannelReference(receivertesting.ChannelName, receivertesting.ChannelNamespace)
 	bindingMessage := receivertesting.CreateBindingMessage(cloudevents.VersionV1)
@@ -111,7 +111,7 @@ func TestConfigChanged(t *testing.T) {
 	commontesting.SetTestEnvironment(t)
 
 	// Test Data
-	brokers := []string{receivertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).Build(ctx)
 	assert.Nil(t, err)
 
@@ -203,11 +203,119 @@ func TestConfigChanged(t *testing.T) {
 	}
 }
 
+// Test The Producer's ConfigChanged Functionality
+func TestSecretChanged(t *testing.T) {
+
+	logger := logtesting.TestLogger(t)
+	ctx := logging.WithLogger(context.TODO(), logger)
+
+	// Setup Test Environment Namespaces
+	commontesting.SetTestEnvironment(t)
+
+	// Test Data
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
+	auth := &commonclient.KafkaAuthConfig{
+		SASL: &commonclient.KafkaSaslConfig{
+			User:     commonconfigtesting.DefaultSecretUsername,
+			Password: commonconfigtesting.DefaultSecretPassword,
+			SaslType: commonconfigtesting.DefaultSecretSaslType,
+		},
+	}
+	baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(commonconfigtesting.DefaultSaramaConfigYaml).WithAuth(auth).Build(ctx)
+	assert.Nil(t, err)
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		only              bool
+		name              string
+		newSecret         *corev1.Secret
+		expectNewProducer bool
+	}
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{
+			name:              "No Changes (Same Producer)",
+			newSecret:         configtesting.NewKafkaSecret(),
+			expectNewProducer: false,
+		},
+		{
+			name:              "Password Change (New Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedPassword),
+			expectNewProducer: true,
+		},
+		{
+			name:              "Username Change (New Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedUsername),
+			expectNewProducer: true,
+		},
+		{
+			name:              "Empty Username Change (New Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithEmptyUsername),
+			expectNewProducer: true,
+		},
+		{
+			name:              "SaslType Change (New Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedSaslType),
+			expectNewProducer: true,
+		},
+		{
+			name:              "Namespace Change (Same Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedNamespace),
+			expectNewProducer: false,
+		},
+		{
+			name:              "No Auth Config In Secret (Same Producer)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithMissingConfig),
+			expectNewProducer: false,
+		},
+	}
+
+	// Filter To Those With "only" Flag (If Any Specified)
+	filteredTestCases := make([]TestCase, 0)
+	for _, testCase := range testCases {
+		if testCase.only {
+			filteredTestCases = append(filteredTestCases, testCase)
+		}
+	}
+	if len(filteredTestCases) == 0 {
+		filteredTestCases = testCases
+	}
+
+	// Make Sure To Restore The NewSyncProducer Wrapper After The Test
+	defer producertesting.RestoreNewSyncProducerFn()
+
+	// Run The Filtered TestCases
+	for _, testCase := range filteredTestCases {
+		t.Run(testCase.name, func(t *testing.T) {
+
+			// Mock The SyncProducer & Stub The NewSyncProducerWrapper()
+			mockSyncProducer := producertesting.NewMockSyncProducer()
+			producertesting.StubNewSyncProducerFn(producertesting.NonValidatingNewSyncProducerFn(mockSyncProducer))
+
+			// Create A Test Producer To Perform Tests Against
+			producer := createTestProducer(t, brokers, baseSaramaConfig, mockSyncProducer)
+
+			// Perform The Test
+			newProducer := producer.SecretChanged(ctx, testCase.newSecret)
+
+			// Verify Expected State
+			assert.Equal(t, testCase.expectNewProducer, newProducer != nil)
+			assert.Equal(t, testCase.expectNewProducer, mockSyncProducer.Closed())
+			if newProducer != nil {
+				assert.Equal(t, producer.brokers, newProducer.brokers)
+				assert.Equal(t, producer.statsReporter, newProducer.statsReporter)
+				assert.Equal(t, producer.healthServer, newProducer.healthServer)
+			}
+		})
+	}
+}
+
 // Test The Producer's Close() Functionality
 func TestClose(t *testing.T) {
 
 	// Test Data
-	brokers := []string{receivertesting.KafkaBroker}
+	brokers := []string{commonconfigtesting.DefaultKafkaBroker}
 	config := sarama.NewConfig()
 
 	// Create A Mock Kafka SyncProducer
diff --git a/pkg/channel/distributed/receiver/testing/constants.go b/pkg/channel/distributed/receiver/testing/constants.go
index fa265e4d4..66a691bc9 100644
--- a/pkg/channel/distributed/receiver/testing/constants.go
+++ b/pkg/channel/distributed/receiver/testing/constants.go
@@ -18,8 +18,6 @@ package testing
 
 // Test Data
 const (
-	KafkaBroker = "TestBroker"
-
 	ChannelName      = "TestChannelName"
 	ChannelNamespace = "TestChannelNamespace"
 
diff --git a/pkg/common/client/config_test.go b/pkg/common/client/config_test.go
index ff5d81746..2d4ac0be1 100644
--- a/pkg/common/client/config_test.go
+++ b/pkg/common/client/config_test.go
@@ -666,6 +666,51 @@ func TestVerifyCertSkipHostname(t *testing.T) {
 	}
 }
 
+func TestHasSameSettings(t *testing.T) {
+	authConfig := &KafkaAuthConfig{SASL: &KafkaSaslConfig{User: "user1", Password: "password1", SaslType: sarama.SASLTypeOAuth}}
+	saramaConfig := sarama.NewConfig()
+
+	assert.False(t, authConfig.SASL.HasSameSettings(saramaConfig))
+	saramaConfig.Net.SASL.User = "user1"
+	assert.False(t, authConfig.SASL.HasSameSettings(saramaConfig))
+	saramaConfig.Net.SASL.Password = "password1"
+	assert.False(t, authConfig.SASL.HasSameSettings(saramaConfig))
+	saramaConfig.Net.SASL.Mechanism = sarama.SASLTypeOAuth
+	assert.True(t, authConfig.SASL.HasSameSettings(saramaConfig))
+}
+
+func TestHasSameBrokers(t *testing.T) {
+
+	// Define The TestCase Struct
+	type TestCase struct {
+		name  string
+		str   string
+		slice []string
+		same  bool
+	}
+
+	const brokers123 = "broker1,broker2,broker3"
+
+	// Create The TestCases
+	testCases := []TestCase{
+		{name: "3 vs. Identical Slice", str: brokers123, slice: []string{"broker1", "broker2", "broker3"}, same: true},
+		{name: "3 vs. Empty Slice", str: brokers123, slice: []string{}},
+		{name: "3 vs. 1-element Slice", str: brokers123, slice: []string{"broker1"}},
+		{name: "3 vs. 2-element Slice", str: brokers123, slice: []string{"broker1", "broker2"}},
+		{name: "3 vs. 4-element Slice", str: brokers123, slice: []string{"broker1", "broker2", "broker3", "broker4"}},
+		{name: "3 vs. Reversed Slice", str: brokers123, slice: []string{"broker3", "broker2", "broker1"}},
+		{name: "1 vs. 1-element Slice", str: "broker1", slice: []string{"broker1"}, same: true},
+		{name: "0 vs. Empty Slice", str: "", slice: []string{}, same: true},
+	}
+
+	// Run The TestCases
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			assert.Equal(t, testCase.same, HasSameBrokers(testCase.str, testCase.slice))
+		})
+	}
+}
+
 // Lifted from the RSA path of https://golang.org/src/crypto/tls/generate_cert.go.
 func generateCert(t *testing.T) (string, string) {
 	t.Helper()
diff --git a/test/e2e-common.sh b/test/e2e-common.sh
index b53337e00..afd539e4b 100755
--- a/test/e2e-common.sh
+++ b/test/e2e-common.sh
@@ -59,7 +59,6 @@ readonly CONFIG_TRACING_CONFIG="test/config/config-tracing.yaml"
 # Strimzi Kafka Cluster Brokers URL (base64 encoded value for k8s secret)
 readonly STRIMZI_KAFKA_NAMESPACE="kafka" # Installation Namespace
 readonly STRIMZI_KAFKA_CLUSTER_BROKERS="my-cluster-kafka-bootstrap.kafka.svc:9092"
-readonly STRIMZI_KAFKA_CLUSTER_BROKERS_ENCODED="bXktY2x1c3Rlci1rYWZrYS1ib290c3RyYXAua2Fma2Euc3ZjOjkwOTI=" # Simple base64 encoding of STRIMZI_KAFKA_CLUSTER_BROKERS
 
 # Eventing Kafka main config path from HEAD.
 readonly KAFKA_CRD_CONFIG_TEMPLATE_DIR="./config/channel"
@@ -303,8 +302,8 @@ function install_distributed_channel_crds() {
   cp "${DISTRIBUTED_TEMPLATE_DIR}/"*yaml "${KAFKA_CRD_CONFIG_DIR}"
   sed -i "s/namespace: knative-eventing/namespace: ${SYSTEM_NAMESPACE}/g" "${KAFKA_CRD_CONFIG_DIR}/"*yaml
 
-  # Update The Kafka Secret With Strimzi Kafka Cluster Brokers (No Authentication)
-  sed -i "s/brokers: \"\"/brokers: ${STRIMZI_KAFKA_CLUSTER_BROKERS_ENCODED}/" "${KAFKA_CRD_CONFIG_DIR}/${EVENTING_KAFKA_SECRET_TEMPLATE}"
+  # Update The ConfigMap With Strimzi Kafka Cluster Brokers (No Authentication)
+  sed -i "s/REPLACE_WITH_CLUSTER_URL/${KAFKA_CLUSTER_URL}/" ${KAFKA_CRD_CONFIG_DIR}/${EVENTING_KAFKA_CONFIG_TEMPLATE}
 
   # Update the config-kafka configmap to disable SASL/TLS for the tests
   sed -i '/^ *TLS:/{n;s/true/false/};/^ *SASL:/{n;s/true/false/}' "${KAFKA_CRD_CONFIG_DIR}/${EVENTING_KAFKA_CONFIG_TEMPLATE}"
diff --git a/test/e2e/kafka_source_test.go b/test/e2e/kafka_source_test.go
index 8f3f385bf..d37128244 100644
--- a/test/e2e/kafka_source_test.go
+++ b/test/e2e/kafka_source_test.go
@@ -422,3 +422,163 @@ func mustJsonMarshal(t *testing.T, val interface{}) string {
 	}
 	return string(data)
 }
+
+type message struct {
+	cloudEventType string
+	payload        []byte
+	headers        map[string]string
+	key            string
+}
+
+type updateTest struct {
+	auth      authSetup
+	topicName string
+	sinkName  string
+}
+
+var (
+	defaultKafkaSource = updateTest{
+		auth: authSetup{
+			bootStrapServer: kafkaBootstrapUrlTLS,
+			SASLEnabled:     false,
+			TLSEnabled:      true,
+		},
+		topicName: "initial-topic",
+		sinkName:  "default-event-recorder",
+	}
+)
+
+func TestKafkaSourceUpdate(t *testing.T) {
+
+	testCases := map[string]updateTest{
+		"no-change": defaultKafkaSource,
+		"change-sink": {
+			auth:      defaultKafkaSource.auth,
+			topicName: defaultKafkaSource.topicName,
+			sinkName:  "update-event-recorder",
+		},
+		"change-topic": {
+			auth:      defaultKafkaSource.auth,
+			topicName: "update-topic",
+			sinkName:  defaultKafkaSource.sinkName,
+		},
+		"change-bootstrap-server": {
+			auth: authSetup{
+				bootStrapServer: kafkaBootstrapUrlPlain,
+				TLSEnabled:      false,
+				SASLEnabled:     false,
+			},
+			topicName: defaultKafkaSource.topicName,
+			sinkName:  defaultKafkaSource.sinkName,
+		},
+	}
+
+	for name, test := range testCases {
+		test := test
+		t.Run(name, func(t *testing.T) {
+			testKafkaSourceUpdate(t, name, test)
+		})
+	}
+}
+
+func testKafkaSourceUpdate(t *testing.T, name string, test updateTest) {
+
+	messagePayload := []byte(`{"value":5}`)
+	matcherGen := func(cloudEventsSourceName, originalOrUpdate string) EventMatcher {
+		return AllOf(
+			HasSource(cloudEventsSourceName),
+			HasData(messagePayload),
+			HasExtension("kafkaheaderceoriginalorupdate", originalOrUpdate))
+	}
+
+	originalMessage := message{
+		payload: messagePayload,
+		headers: map[string]string{
+			"ce_originalorupdate": "original",
+		},
+		key: "0",
+	}
+	updateMessage := message{
+		payload: messagePayload,
+		headers: map[string]string{
+			"ce_originalorupdate": "update",
+		},
+		key: "0",
+	}
+
+	client := testlib.Setup(t, true)
+	defer testlib.TearDown(client)
+
+	t.Logf("Creating topic: %s\n", defaultKafkaSource.topicName+name)
+	helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, defaultKafkaSource.topicName+name)
+
+	t.Logf("Copying secrets: %s\n", defaultKafkaSource.topicName+name)
+	_, err := utils.CopySecret(client.Kube.CoreV1(), "knative-eventing", kafkaTLSSecret, client.Namespace, "default")
+	if err != nil {
+		t.Fatalf("could not copy secret(%s): %v", kafkaTLSSecret, err)
+	}
+
+	t.Logf("Creating default eventrecorder pod: %s\n", defaultKafkaSource.sinkName)
+	originalEventTracker, _ := recordevents.StartEventRecordOrFail(context.Background(), client, defaultKafkaSource.sinkName)
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	kafkaSourceName := "e2e-kafka-source-" + name
+
+	t.Logf("Creating kafkasource: %s\n", kafkaSourceName)
+	contribtestlib.CreateKafkaSourceV1Beta1OrFail(client, contribresources.KafkaSourceV1Beta1(
+		defaultKafkaSource.auth.bootStrapServer,
+		defaultKafkaSource.topicName+name,
+		resources.ServiceRef(defaultKafkaSource.sinkName),
+		contribresources.WithNameV1Beta1(kafkaSourceName),
+		withAuthEnablementV1Beta1(defaultKafkaSource.auth),
+	))
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	t.Logf("Send update event to kafkatopic")
+	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
+		defaultKafkaSource.topicName+name,
+		originalMessage.key, originalMessage.headers, string(originalMessage.payload))
+	eventSourceName := sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, defaultKafkaSource.topicName+name)
+	originalEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "original")))
+	t.Logf("Properly received original event for %s\n", eventSourceName)
+
+	var (
+		ksObj               *sourcesv1beta1.KafkaSource
+		newSinkEventTracker *recordevents.EventInfoStore
+	)
+	ksObj = contribtestlib.GetKafkaSourceV1Beta1OrFail(client, kafkaSourceName)
+	if ksObj == nil {
+		t.Fatalf("Unabled to Get kafkasource: %s/%s\n", client.Namespace, kafkaSourceName)
+	}
+	if test.topicName != defaultKafkaSource.topicName {
+		helpers.MustCreateTopic(client, kafkaClusterName, kafkaClusterNamespace, test.topicName+name)
+		ksObj.Spec.Topics = []string{test.topicName + name}
+		eventSourceName = sourcesv1beta1.KafkaEventSource(client.Namespace, kafkaSourceName, test.topicName+name)
+	}
+	if test.sinkName != defaultKafkaSource.sinkName {
+		kSinkRef := resources.ServiceKRef(test.sinkName)
+		ksObj.Spec.Sink.Ref = kSinkRef
+
+		newSinkEventTracker, _ = recordevents.StartEventRecordOrFail(context.Background(), client, test.sinkName)
+	}
+	if test.auth.bootStrapServer != defaultKafkaSource.auth.bootStrapServer {
+		ksObj.Spec.KafkaAuthSpec.BootstrapServers = []string{test.auth.bootStrapServer}
+		ksObj.Spec.KafkaAuthSpec.Net.TLS.Enable = test.auth.TLSEnabled
+		ksObj.Spec.KafkaAuthSpec.Net.SASL.Enable = test.auth.SASLEnabled
+	}
+
+	contribtestlib.UpdateKafkaSourceV1Beta1OrFail(client, ksObj)
+	client.WaitForAllTestResourcesReadyOrFail(context.Background())
+
+	t.Logf("Send update event to kafkatopic")
+	helpers.MustPublishKafkaMessage(client, kafkaBootstrapUrlPlain,
+		test.topicName+name,
+		updateMessage.key, updateMessage.headers, string(updateMessage.payload))
+
+	if test.sinkName != defaultKafkaSource.sinkName {
+		newSinkEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "update")))
+	} else {
+		originalEventTracker.AssertExact(1, recordevents.MatchEvent(matcherGen(eventSourceName, "update")))
+	}
+
+}
diff --git a/test/lib/creation.go b/test/lib/creation.go
index bddcf311b..01e281d2f 100644
--- a/test/lib/creation.go
+++ b/test/lib/creation.go
@@ -58,6 +58,33 @@ func CreateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1bet
 	}
 }
 
+func GetKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource string) *sourcesv1beta1.KafkaSource {
+	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+	}
+
+	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+	if ksObj, err := kSources.Get(context.Background(), kafkaSource, metav1.GetOptions{}); err != nil {
+		c.T.Fatalf("Failed to get v1beta1 KafkaSource %q: %v", kafkaSource, err)
+	} else {
+		return ksObj
+	}
+	return nil
+}
+
+func UpdateKafkaSourceV1Beta1OrFail(c *testlib.Client, kafkaSource *sourcesv1beta1.KafkaSource) {
+	kafkaSourceClientSet, err := kafkaclientset.NewForConfig(c.Config)
+	if err != nil {
+		c.T.Fatalf("Failed to create v1beta1 KafkaSource client: %v", err)
+	}
+
+	kSources := kafkaSourceClientSet.SourcesV1beta1().KafkaSources(c.Namespace)
+	if _, err := kSources.Update(context.Background(), kafkaSource, metav1.UpdateOptions{}); err != nil {
+		c.T.Fatalf("Failed to update v1beta1 KafkaSource %q: %v", kafkaSource.Name, err)
+	}
+}
+
 func CreateKafkaBindingV1Alpha1OrFail(c *testlib.Client, kafkaBinding *bindingsv1alpha1.KafkaBinding) {
 	kafkaBindingClientSet, err := kafkaclientset.NewForConfig(c.Config)
 	if err != nil {