Repository: knative-sandbox/eventing-kafka
Pull request #696 (2021-06-08 20:39:04)

do not print out  users' apikey
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 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

```

**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/696#" title="Author self-approved">steven0711dong</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)~~ [steven0711dong]

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":[]} -->
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/common/client/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/696/pull-knative-sandbox-eventing-kafka-go-coverage/1402364585972862976/artifacts/line-cov.html#file0) | 91.7% | 91.5% | -0.2

knative-prow-robot: @steven0711dong: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-upgrade-tests | a0a55b363b1680202887042260f9b0a94199d743 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/696/pull-knative-sandbox-eventing-kafka-upgrade-tests/1402364585880588288) | `/test pull-knative-sandbox-eventing-kafka-upgrade-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka/696). [Your PR dashboard](https://gubernator.knative.dev/pr/steven0711dong).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
lionelvillard: /lgtm
knative-prow-robot: New changes are detected. LGTM label has been removed.
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/common/client/config.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/696/pull-knative-sandbox-eventing-kafka-go-coverage/1402679667852316672/artifacts/line-cov.html#file0) | 91.7% | 91.7% | 0.1

lionelvillard: /lgtm

/hold for @aliok @matzew to have a chance to TAL.
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#696](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0a29438) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/24bc388317d4687dcf616377a333f3932d0aab7c?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (24bc388) will **decrease** coverage by `0.49%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/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/696?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     #696      +/-   ##
==========================================
- Coverage   75.17%   74.67%   -0.50%     
==========================================
  Files         132      132              
  Lines        5784     5817      +33     
==========================================
- Hits         4348     4344       -4     
- Misses       1215     1251      +36     
- Partials      221      222       +1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/common/client/config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NvbW1vbi9jbGllbnQvY29uZmlnLmdv) | `89.56% <100.00%> (+0.09%)` | :arrow_up: |
| [pkg/source/mtadapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9tdGFkYXB0ZXIvYWRhcHRlci5nbw==) | `56.41% <0.00%> (-18.78%)` | :arrow_down: |
| [...onsolidated/dispatcher/consumer\_message\_handler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvY29uc3VtZXJfbWVzc2FnZV9oYW5kbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [pkg/channel/consolidated/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvY29uc29saWRhdGVkL2Rpc3BhdGNoZXIvZGlzcGF0Y2hlci5nbw==) | `61.58% <0.00%> (+0.10%)` | :arrow_up: |
| [pkg/source/adapter/adapter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9hZGFwdGVyL2FkYXB0ZXIuZ28=) | `59.50% <0.00%> (+0.33%)` | :arrow_up: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `80.53% <0.00%> (+6.19%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696?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/696?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [2f12463...0a29438](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/696?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).


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
-}
diff --git a/pkg/common/client/config.go b/pkg/common/client/config.go
index 7ff20dd84..caf3359bb 100644
--- a/pkg/common/client/config.go
+++ b/pkg/common/client/config.go
@@ -253,7 +253,6 @@ func (b *configBuilder) Build(ctx context.Context) (*sarama.Config, error) {
 				config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
 			}
 			config.Net.SASL.User = b.auth.SASL.User
-			config.Net.SASL.Password = b.auth.SASL.Password
 		}
 	}
 
@@ -268,6 +267,10 @@ func (b *configBuilder) Build(ctx context.Context) (*sarama.Config, error) {
 	logger := logging.FromContext(ctx)
 	logger.Infof("Built Sarama config: %+v", config)
 
+	if b.auth != nil && b.auth.SASL != nil {
+		config.Net.SASL.Password = b.auth.SASL.Password
+	}
+
 	return config, nil
 }
 


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