Repository: knative-sandbox/eventing-kafka
Pull request #700 (2021-06-09 22:15:50)

Remove creation of new dispatcher when secret changes
This isn't directly tied to an issue, but the changes for #537 made a few things in the distributed channel's "reconfigure" function obsolete; in particular it used to be called when the configmap changed as well, which could require a new dispatcher.  Now the only possible changes are from the secret, which is limited to username/password/SASLtype.  Changing those settings does not require a new dispatcher, so this PR will remove that behavior.

I would leave this alone but there is also a bug in the way the new dispatcher is created which leaves the old dispatcher referenced in the dispatcher's controller.  The controller.NewController call uses the original dispatcher and nothing ever changes that, which is something that will cause a problem with some upcoming changes that include a "consumer group manager" that maintains some state information about consumer groups.  This manager can't have "a different manager" in an abandoned dispatcher elsewhere, so it needs to be fixed.

## Proposed Changes

- 🧽  Changing the kafka-cluster secret no longer creates a new distributed dispatcher; instead simply reconfigures the existing one
- 🐛 Removing (vs. changing) authentication by changing the kafka-cluster secret had no effect; now it removes the auth properly (dispatcher and receiver)
- 🧽  merged the "reconfigure" functions from dispatcher and receiver into the SecretChanged functions (since they were only being called once now that there is no longer ConfigMap-watching functionality in these components)


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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["davyodom"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/700/pull-knative-sandbox-eventing-kafka-go-coverage/1402751324902133760/artifacts/line-cov.html#file1) | 82.5% | 81.2% | -1.3
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/700/pull-knative-sandbox-eventing-kafka-go-coverage/1402751324902133760/artifacts/line-cov.html#file2) | 76.7% | 78.1% | 1.4

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#700](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c31c6f7) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/f95e2c1c5fa237cef1389e8e64a53dbf93444979?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f95e2c1) will **increase** coverage by `0.00%`.
> The diff coverage is `70.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@           Coverage Diff           @@
##             main     #700   +/-   ##
=======================================
  Coverage   74.55%   74.55%           
=======================================
  Files         132      132           
  Lines        5816     5813    -3     
=======================================
- Hits         4336     4334    -2     
  Misses       1258     1258           
+ Partials      222      221    -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `72.72% <70.00%> (-1.61%)` | :arrow_down: |
| [.../channel/distributed/receiver/producer/producer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/700/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvcmVjZWl2ZXIvcHJvZHVjZXIvcHJvZHVjZXIuZ28=) | `69.31% <70.00%> (+2.27%)` | :arrow_up: |

------

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

travis-minke-sap: Looks good to me - leaving open for others to review.  Changes are primarily in distributed channel with only a small change in "common" code.  I will likely merge Thursday 6/10 morning US time so get any review comments in prior to that ; )
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/700/pull-knative-sandbox-eventing-kafka-go-coverage/1402758125353701376/artifacts/line-cov.html#file1) | 82.5% | 82.2% | -0.3
[pkg/channel/distributed/receiver/producer/producer.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/700/pull-knative-sandbox-eventing-kafka-go-coverage/1402758125353701376/artifacts/line-cov.html#file2) | 76.7% | 78.1% | 1.4

travis-minke-sap: /approve
/lgtm
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/700#" title="Author self-approved">eric-sap</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/700#issuecomment-858653140" title="LGTM">travis-minke-sap</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/main/OWNERS)~~ [travis-minke-sap]

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":[]} -->

Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index b1feec31f..d71ea2578 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -209,24 +209,19 @@ func flush(logger *zap.Logger) {
 
 // secretObserver is the callback function that handles changes to our Secret
 func secretObserver(ctx context.Context, secret *corev1.Secret) {
-	logger := logging.FromContext(ctx)
+	secretLogger := logging.FromContext(ctx)
 
 	if secret == nil {
-		logger.Warn("Nil Secret passed to secretObserver; ignoring")
+		secretLogger.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")
+		secretLogger.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
-	}
+	dispatcher.SecretChanged(ctx, secret)
 }
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 6e8a1821d..012014a32 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -18,7 +18,6 @@ package dispatcher
 
 import (
 	"context"
-	"strings"
 	"sync"
 	"time"
 
@@ -66,7 +65,7 @@ func NewSubscriberWrapper(subscriberSpec eventingduck.SubscriberSpec, groupId st
 
 // Dispatcher Interface
 type Dispatcher interface {
-	SecretChanged(ctx context.Context, secret *corev1.Secret) Dispatcher
+	SecretChanged(ctx context.Context, secret *corev1.Secret)
 	Shutdown()
 	UpdateSubscriptions(subscriberSpecs []eventingduck.SubscriberSpec) map[eventingduck.SubscriberSpec]error
 }
@@ -276,8 +275,8 @@ func (d *DispatcherImpl) closeConsumerGroup(subscriber *SubscriberWrapper) {
 }
 
 // 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 {
+// settings specific to the dispatcher may be changed if necessary
+func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secret) {
 
 	// Debug Log The Secret Change
 	d.Logger.Debug("New Secret Received", zap.String("secret.Name", secret.ObjectMeta.Name))
@@ -285,46 +284,47 @@ func (d *DispatcherImpl) SecretChanged(ctx context.Context, secret *corev1.Secre
 	kafkaAuthCfg := commonconfig.GetAuthConfigFromSecret(secret)
 	if kafkaAuthCfg == nil {
 		d.Logger.Warn("No auth config found in secret; ignoring update")
-		return nil
+		return
 	}
 
 	// 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
+		return
 	}
 
 	// 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
+		// Any existing SASL config must be cleared explicitly or the "WithExisting" builder will keep the old values
+		d.SaramaConfig.Net.SASL.Enable = false
+		d.SaramaConfig.Net.SASL.User = ""
+		d.SaramaConfig.Net.SASL.Password = ""
 	}
 	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
+		return
 	}
 
 	// 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)
-}
 
-// reconfigure shuts down the current dispatcher and recreates it with new settings
-func (d *DispatcherImpl) reconfigure(newConfig *sarama.Config, ekConfig *commonconfig.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, ",")
+	// Close all of the ConsumerGroups so that the settings can be changed
+	// Note: Not calling d.Shutdown because we don't want to close the metrics channel here
+	for _, subscriber := range d.subscribers {
+		d.closeConsumerGroup(subscriber)
 	}
-	newDispatcher := NewDispatcher(d.DispatcherConfig)
-	failedSubscriptions := newDispatcher.UpdateSubscriptions(d.SubscriberSpecs)
+
+	// The only values in the secret that matter here are the username, password, and SASL type, which are all part
+	// of the SaramaConfig, so that's all that needs to be modified
+	d.DispatcherConfig.SaramaConfig = newConfig
+
+	failedSubscriptions := d.UpdateSubscriptions(d.SubscriberSpecs)
 	if len(failedSubscriptions) > 0 {
-		d.Logger.Fatal("Failed To Subscribe Kafka Subscriptions For New Dispatcher", zap.Int("Count", len(failedSubscriptions)))
-		return nil
+		d.Logger.Fatal("Failed To Subscribe Kafka Subscriptions Using Updated Secret", zap.Int("Count", len(failedSubscriptions)))
 	}
-	return newDispatcher
 }
 
 // ObserveMetrics Is An Async Process For Observing Kafka Metrics
diff --git a/pkg/channel/distributed/receiver/producer/producer.go b/pkg/channel/distributed/receiver/producer/producer.go
index 549c86f48..e454d3c1d 100644
--- a/pkg/channel/distributed/receiver/producer/producer.go
+++ b/pkg/channel/distributed/receiver/producer/producer.go
@@ -19,7 +19,6 @@ package producer
 import (
 	"context"
 	"errors"
-	"strings"
 	"time"
 
 	"github.com/Shopify/sarama"
@@ -195,6 +194,10 @@ func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Pr
 	if kafkaAuthCfg.SASL.User == "" {
 		// The config builder expects the entire config object to be nil if not using auth
 		kafkaAuthCfg = nil
+		// Any existing SASL config must be cleared explicitly or the "WithExisting" builder will keep the old values
+		p.configuration.Net.SASL.Enable = false
+		p.configuration.Net.SASL.User = ""
+		p.configuration.Net.SASL.Password = ""
 	}
 	newConfig, err := client.NewConfigBuilder().WithExisting(p.configuration).WithAuth(kafkaAuthCfg).Build(ctx)
 	if err != nil {
@@ -204,7 +207,18 @@ func (p *Producer) SecretChanged(ctx context.Context, secret *corev1.Secret) *Pr
 
 	// 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)
+
+	// Shut down the current producer and recreate it with new settings
+	p.Close()
+	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
 }
 
 // Close The Producer (Stop Processing)
@@ -225,21 +239,3 @@ 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 *commonconfig.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
-}


Test patch:
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index a13740148..e2adbbe17 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -218,6 +218,4 @@ func (m MockDispatcher) UpdateSubscriptions(_ []eventingduck.SubscriberSpec) map
 	return nil
 }
 
-func (m MockDispatcher) SecretChanged(_ context.Context, _ *corev1.Secret) dispatcher.Dispatcher {
-	return nil
-}
+func (m MockDispatcher) SecretChanged(_ context.Context, _ *corev1.Secret) {}
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index 28c089944..283713074 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -71,8 +71,14 @@ func TestNewSubscriberWrapper(t *testing.T) {
 // Test The NewDispatcher() Functionality
 func TestNewDispatcher(t *testing.T) {
 
+	baseSaramaConfig, err := commonclient.NewConfigBuilder().
+		WithDefaults().
+		WithVersion(&sarama.V2_0_0_0).
+		Build(context.TODO())
+	assert.Nil(t, err)
+
 	// Perform The Test & Verify Results (Not Much To See Due To Interface)
-	createTestDispatcher(t, nil, nil)
+	createTestDispatcher(t, nil, baseSaramaConfig)
 }
 
 // Test The Dispatcher's Shutdown() Functionality
@@ -312,67 +318,55 @@ func TestSecretChanged(t *testing.T) {
 
 	// Define The TestCase Struct
 	type TestCase struct {
-		only                bool
 		name                string
 		newSecret           *corev1.Secret
-		expectNewDispatcher bool
+		expectEmptyUsername bool
+		expectNewUsername   string
+		expectNewPassword   string
+		expectNewSaslType   string
 	}
 
 	// Create The TestCases
 	testCases := []TestCase{
 		{
-			name:                "No Changes (Same Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(),
-			expectNewDispatcher: false,
+			name:      "No Changes (No Modifications)",
+			newSecret: configtesting.NewKafkaSecret(),
 		},
 		{
-			name:                "Password Change (New Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedPassword),
-			expectNewDispatcher: true,
+			name:              "Password Change (Modifications)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedPassword),
+			expectNewPassword: configtesting.ModifiedSecretPassword,
 		},
 		{
-			name:                "Username Change (New Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedUsername),
-			expectNewDispatcher: true,
+			name:              "Username Change (Modifications)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedUsername),
+			expectNewUsername: configtesting.ModifiedSecretUsername,
 		},
 		{
-			name:                "Empty Username Change (New Dispatcher)",
+			name:                "Empty Username Change (Modifications)",
 			newSecret:           configtesting.NewKafkaSecret(configtesting.WithEmptyUsername),
-			expectNewDispatcher: true,
+			expectEmptyUsername: true,
 		},
 		{
-			name:                "SaslType Change (New Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedSaslType),
-			expectNewDispatcher: true,
+			name:              "SaslType Change (Modifications)",
+			newSecret:         configtesting.NewKafkaSecret(configtesting.WithModifiedSaslType),
+			expectNewSaslType: configtesting.ModifiedSecretSaslType,
 		},
 		{
-			name:                "Namespace Change (Same Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(configtesting.WithModifiedNamespace),
-			expectNewDispatcher: false,
+			name:      "Namespace Change (No Modifications)",
+			newSecret: configtesting.NewKafkaSecret(configtesting.WithModifiedNamespace),
 		},
 		{
-			name:                "No Auth Config In Secret (Same Dispatcher)",
-			newSecret:           configtesting.NewKafkaSecret(configtesting.WithMissingConfig),
-			expectNewDispatcher: false,
+			name:      "No Auth Config In Secret (No Modifications)",
+			newSecret: configtesting.NewKafkaSecret(configtesting.WithMissingConfig),
 		},
 	}
 
-	// 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 {
+	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
 			// Mock The SyncProducer & Stub The NewConsumerGroupWrapper()
@@ -381,19 +375,26 @@ func TestSecretChanged(t *testing.T) {
 
 			// Create A Test Dispatcher To Perform Tests Against
 			dispatcher := createTestDispatcher(t, brokers, baseSaramaConfig)
+			impl := dispatcher.(*DispatcherImpl)
+			impl.subscribers = map[types.UID]*SubscriberWrapper{uid123: createSubscriberWrapper(uid123)}
 
 			// 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)
-
-			if testCase.expectNewDispatcher {
-				// Verify that the new dispatcher's channels are not the same as the original
-				oldImpl := dispatcher.(*DispatcherImpl)
-				newImpl := newDispatcher.(*DispatcherImpl)
-				assert.NotEqual(t, oldImpl.MetricsStopChan, newImpl.MetricsStopChan)
-				assert.NotEqual(t, oldImpl.MetricsStoppedChan, newImpl.MetricsStoppedChan)
+			dispatcher.SecretChanged(ctx, testCase.newSecret)
+			assert.NotNil(t, impl)
+
+			if testCase.expectEmptyUsername {
+				// An empty username in the secret will force no-authorization even if it was enabled before
+				assert.Equal(t, false, impl.SaramaConfig.Net.SASL.Enable)
+				assert.Equal(t, "", impl.SaramaConfig.Net.SASL.User)
+				assert.Equal(t, "", impl.SaramaConfig.Net.SASL.Password)
+			} else if testCase.expectNewUsername != "" {
+				assert.Equal(t, testCase.expectNewUsername, impl.SaramaConfig.Net.SASL.User)
+			}
+			if testCase.expectNewPassword != "" {
+				assert.Equal(t, testCase.expectNewPassword, impl.SaramaConfig.Net.SASL.Password)
+			}
+			if testCase.expectNewSaslType != "" {
+				assert.Equal(t, testCase.expectNewSaslType, string(impl.SaramaConfig.Net.SASL.Mechanism))
 			}
 		})
 	}
@@ -421,6 +422,7 @@ func createTestDispatcher(t *testing.T, brokers []string, config *sarama.Config)
 		Logger:          logger,
 		Brokers:         brokers,
 		StatsReporter:   statsReporter,
+		MetricsRegistry: config.MetricRegistry,
 		SaramaConfig:    config,
 		SubscriberSpecs: subscriberSpecs,
 	}
diff --git a/pkg/channel/distributed/receiver/producer/producer_test.go b/pkg/channel/distributed/receiver/producer/producer_test.go
index 5272b6f84..666ec48a5 100644
--- a/pkg/channel/distributed/receiver/producer/producer_test.go
+++ b/pkg/channel/distributed/receiver/producer/producer_test.go
@@ -119,15 +119,13 @@ func TestSecretChanged(t *testing.T) {
 			SaslType: configtesting.DefaultSecretSaslType,
 		},
 	}
-	baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(clienttesting.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
+		expectEmptyAuth   bool
 	}
 
 	// Create The TestCases
@@ -151,6 +149,7 @@ func TestSecretChanged(t *testing.T) {
 			name:              "Empty Username Change (New Producer)",
 			newSecret:         configtesting.NewKafkaSecret(configtesting.WithEmptyUsername),
 			expectNewProducer: true,
+			expectEmptyAuth:   true,
 		},
 		{
 			name:              "SaslType Change (New Producer)",
@@ -169,22 +168,11 @@ func TestSecretChanged(t *testing.T) {
 		},
 	}
 
-	// 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 {
+	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 
 			// Mock The SyncProducer & Stub The NewSyncProducerWrapper()
@@ -192,6 +180,8 @@ func TestSecretChanged(t *testing.T) {
 			producertesting.StubNewSyncProducerFn(producertesting.NonValidatingNewSyncProducerFn(mockSyncProducer))
 
 			// Create A Test Producer To Perform Tests Against
+			baseSaramaConfig, err := commonclient.NewConfigBuilder().WithDefaults().FromYaml(clienttesting.DefaultSaramaConfigYaml).WithAuth(auth).Build(ctx)
+			assert.Nil(t, err)
 			producer := createTestProducer(t, brokers, baseSaramaConfig, mockSyncProducer)
 
 			// Perform The Test
@@ -201,6 +191,12 @@ func TestSecretChanged(t *testing.T) {
 			assert.Equal(t, testCase.expectNewProducer, newProducer != nil)
 			assert.Equal(t, testCase.expectNewProducer, mockSyncProducer.Closed())
 			if newProducer != nil {
+				if testCase.expectEmptyAuth {
+					// An empty username in the secret will force no-authorization even if it was enabled before
+					assert.Equal(t, false, newProducer.configuration.Net.SASL.Enable)
+					assert.Equal(t, "", newProducer.configuration.Net.SASL.User)
+					assert.Equal(t, "", newProducer.configuration.Net.SASL.Password)
+				}
 				assert.Equal(t, producer.brokers, newProducer.brokers)
 				assert.Equal(t, producer.statsReporter, newProducer.statsReporter)
 				assert.Equal(t, producer.healthServer, newProducer.healthServer)
diff --git a/pkg/common/config/testing/util.go b/pkg/common/config/testing/util.go
index f62630e45..7f0e0ccf7 100644
--- a/pkg/common/config/testing/util.go
+++ b/pkg/common/config/testing/util.go
@@ -17,11 +17,13 @@ limitations under the License.
 package testing
 
 import (
+	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/system"
+
 	commonconstants "knative.dev/eventing-kafka/pkg/common/constants"
 	commontesting "knative.dev/eventing-kafka/pkg/common/testing"
-	"knative.dev/pkg/system"
 )
 
 // Constants
@@ -29,8 +31,13 @@ const (
 	DefaultKafkaBroker     = "TestBroker"
 	DefaultSecretUsername  = "TestUsername"
 	DefaultSecretPassword  = "TestPassword"
-	DefaultSecretSaslType  = "PLAIN"
+	DefaultSecretSaslType  = sarama.SASLTypePlaintext
 	DefaultSecretNamespace = "TestNamespace"
+
+	ModifiedSecretUsername  = "TestModifiedUsername"
+	ModifiedSecretPassword  = "TestModifiedPassword"
+	ModifiedSecretSaslType  = sarama.SASLTypeSCRAMSHA256
+	ModifiedSecretNamespace = "TestModifiedNamespace"
 )
 
 // KafkaSecretOption Enables Customization Of An Eventing-Kafka Secret
@@ -68,12 +75,12 @@ func NewKafkaSecret(options ...KafkaSecretOption) *corev1.Secret {
 
 // WithModifiedPassword Modifies The Default Password Section Of The Secret Data
 func WithModifiedPassword(secret *corev1.Secret) {
-	secret.Data[commonconstants.KafkaSecretKeyPassword] = []byte("TestModifiedPassword")
+	secret.Data[commonconstants.KafkaSecretKeyPassword] = []byte(ModifiedSecretPassword)
 }
 
 // WithModifiedUsername Modifies The Default Username Section Of The Secret Data
 func WithModifiedUsername(secret *corev1.Secret) {
-	secret.Data[commonconstants.KafkaSecretKeyUsername] = []byte("TestModifiedUsername")
+	secret.Data[commonconstants.KafkaSecretKeyUsername] = []byte(ModifiedSecretUsername)
 }
 
 // WithEmptyUsername Empties The Default Username Section Of The Secret Data
@@ -83,12 +90,12 @@ func WithEmptyUsername(secret *corev1.Secret) {
 
 // WithModifiedSaslType Modifies The Default SaslType Section Of The Secret Data
 func WithModifiedSaslType(secret *corev1.Secret) {
-	secret.Data[commonconstants.KafkaSecretKeySaslType] = []byte("TestModifiedSaslType")
+	secret.Data[commonconstants.KafkaSecretKeySaslType] = []byte(ModifiedSecretSaslType)
 }
 
 // WithModifiedNamespace Modifies The Default Namespace Section Of The Secret Data
 func WithModifiedNamespace(secret *corev1.Secret) {
-	secret.Data[commonconstants.KafkaSecretKeyNamespace] = []byte("TestModifiedNamespace")
+	secret.Data[commonconstants.KafkaSecretKeyNamespace] = []byte(ModifiedSecretNamespace)
 }
 
 // WithMissingConfig Removes the Data From The Secret