Repository: knative-sandbox/eventing-kafka
Pull request #489 (2021-03-29 21:25:45)

Add "cloudevents" to the eventing-kafka config
## Proposed Changes

The consolidated channel allows configuration of the cloudevents args (max idle connections, max idle connections per host), but the distributed channel did not have this capability.  This PR will allow configuration of these fields via the config-kafka configmap.  For example:
```
  eventing-kafka: |
    cloudevents:
      maxIdleConns: 50
      maxIdleConnsPerHost: 5
```

If no "cloudevents" field is present in the eventing-kafka YAML string, the original behavior (1000 maxIdleConns, 100 maxIdleConnsPerHost) is maintained, as is true for the consolidated channel equivalents.

knative-prow-robot: Hi @eric-sap. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<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.
</details>

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

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` 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":["aliok"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489?src=pr&el=h1) Report
> Merging [#489](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489?src=pr&el=desc) (cfaf045) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/cf838119992315b00fa559fe40bc8db2dd79debb?el=desc) (cf83811) will **increase** coverage by `0.15%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##             main     #489      +/-   ##
==========================================
+ Coverage   73.08%   73.23%   +0.15%     
==========================================
  Files         139      139              
  Lines        5487     5492       +5     
==========================================
+ Hits         4010     4022      +12     
+ Misses       1229     1222       -7     
  Partials      248      248              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [.../channel/distributed/common/kafka/sarama/sarama.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvY29tbW9uL2thZmthL3NhcmFtYS9zYXJhbWEuZ28=) | `88.67% <100.00%> (+1.17%)` | :arrow_up: |
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `76.80% <0.00%> (+5.59%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489?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/489?src=pr&el=footer). Last update [cf83811...cfaf045](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/489?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

aliok: /ok-to-test
aliok: /lgtm

BTW, this cloud connection arguments thing is one of the reasons I was suggesting we should disable on-the-fly reloading of kafka config since this function can be set once per process.
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/common/kafka/sarama/sarama.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/489/pull-knative-sandbox-eventing-kafka-go-coverage/1376865905715187712/artifacts/line-cov.html#file0) | 92.1% | 93.0% | 0.9

knative-prow-robot: New changes are detected. LGTM label has been removed.
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **aliok** after the PR has been reviewed.

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":["aliok"]} -->
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/common/kafka/sarama/sarama.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/489/pull-knative-sandbox-eventing-kafka-go-coverage/1376895324928348160/artifacts/line-cov.html#file1) | 92.1% | 93.0% | 0.9

knative-prow-robot: @eric-sap: PR needs rebase.

<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.
</details>
matzew: @eric-sap mind rebasing? looks like there is a merge conflict

Otherwise looks good - thanks for moving the files around!
eric-sap: Of course; merge testing should be complete shortly
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/common/kafka/sarama/sarama.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/489/pull-knative-sandbox-eventing-kafka-go-coverage/1377370138268930048/artifacts/line-cov.html#file1) | 92.1% | 93.0% | 0.9


Patch:
diff --git a/cmd/channel/distributed/dispatcher/main.go b/cmd/channel/distributed/dispatcher/main.go
index 2519846f7..3d7666b0b 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -127,26 +127,24 @@ func main() {
 	statsReporter := metrics.NewStatsReporter(logger)
 	defer statsReporter.Shutdown()
 
-	// Increase The Idle Connection Limits From Transport Defaults (see net/http/DefaultTransport)
+	// Change The CloudEvent Connection Args
 	kncloudevents.ConfigureConnectionArgs(&kncloudevents.ConnectionArgs{
-		MaxIdleConns:        constants.DefaultMaxIdleConns,
-		MaxIdleConnsPerHost: constants.DefaultMaxIdleConnsPerHost,
+		MaxIdleConns:        ekConfig.CloudEvents.MaxIdleConns,
+		MaxIdleConnsPerHost: ekConfig.CloudEvents.MaxIdleConnsPerHost,
 	})
 
 	// Create The Dispatcher With Specified Configuration
 	dispatcherConfig := dispatch.DispatcherConfig{
-		Logger:             logger,
-		ClientId:           constants.Component,
-		Brokers:            strings.Split(ekConfig.Kafka.Brokers, ","),
-		Topic:              environment.KafkaTopic,
-		Username:           kafkaAuthCfg.SASL.User,
-		Password:           kafkaAuthCfg.SASL.Password,
-		ChannelKey:         environment.ChannelKey,
-		StatsReporter:      statsReporter,
-		MetricsRegistry:    saramaConfig.MetricRegistry,
-		MetricsStopChan:    make(chan struct{}),
-		MetricsStoppedChan: make(chan struct{}),
-		SaramaConfig:       saramaConfig,
+		Logger:          logger,
+		ClientId:        constants.Component,
+		Brokers:         strings.Split(ekConfig.Kafka.Brokers, ","),
+		Topic:           environment.KafkaTopic,
+		Username:        kafkaAuthCfg.SASL.User,
+		Password:        kafkaAuthCfg.SASL.Password,
+		ChannelKey:      environment.ChannelKey,
+		StatsReporter:   statsReporter,
+		MetricsRegistry: saramaConfig.MetricRegistry,
+		SaramaConfig:    saramaConfig,
 	}
 	dispatcher = dispatch.NewDispatcher(dispatcherConfig)
 
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index 8e8c43ce9..4d3dbad82 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -53,9 +53,6 @@ const (
 	KafkaChannelSeparator = "."
 
 	knativeKafkaTopicPrefix = "knative-messaging-kafka"
-
-	DefaultMaxIdleConns        = 1000
-	DefaultMaxIdleConnsPerHost = 100
 )
 
 type KafkaConfig struct {
@@ -129,8 +126,8 @@ func GetKafkaConfig(configMap map[string]string) (*KafkaConfig, error) {
 	}
 
 	config := &KafkaConfig{
-		MaxIdleConns:        DefaultMaxIdleConns,
-		MaxIdleConnsPerHost: DefaultMaxIdleConnsPerHost,
+		MaxIdleConns:        constants.DefaultMaxIdleConns,
+		MaxIdleConnsPerHost: constants.DefaultMaxIdleConnsPerHost,
 	}
 
 	var bootstrapServers string
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama.go b/pkg/channel/distributed/common/kafka/sarama/sarama.go
index 5dfe1e5e6..fe05d5f17 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama.go
@@ -98,6 +98,18 @@ func LoadEventingKafkaSettings(configMap *corev1.ConfigMap) (*commonconfig.Event
 		return nil, fmt.Errorf("ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: %s : %v", err, configMap.Data[constants.EventingKafkaSettingsConfigKey])
 	}
 
+	if eventingKafkaConfig != nil {
+		// If Any Config Was Provided, Set Some Default Values If Missing
+
+		// Increase The Idle Connection Limits From Transport Defaults If Not Provided (see net/http/DefaultTransport)
+		if eventingKafkaConfig.CloudEvents.MaxIdleConns == 0 {
+			eventingKafkaConfig.CloudEvents.MaxIdleConns = constants.DefaultMaxIdleConns
+		}
+		if eventingKafkaConfig.CloudEvents.MaxIdleConnsPerHost == 0 {
+			eventingKafkaConfig.CloudEvents.MaxIdleConnsPerHost = constants.DefaultMaxIdleConnsPerHost
+		}
+	}
+
 	return eventingKafkaConfig, nil
 }
 
diff --git a/pkg/channel/distributed/dispatcher/constants/constants.go b/pkg/channel/distributed/dispatcher/constants/constants.go
index 1c8e8ae15..280f936c9 100644
--- a/pkg/channel/distributed/dispatcher/constants/constants.go
+++ b/pkg/channel/distributed/dispatcher/constants/constants.go
@@ -20,10 +20,6 @@ import "time"
 
 // Global Constants
 const (
-	// Values Used When Creating A New Dispatcher
-	DefaultMaxIdleConns        = 1000
-	DefaultMaxIdleConnsPerHost = 100
-
 	MetricsInterval = 5 * time.Second
 
 	Component = "eventing-kafka-channel-dispatcher"
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
index 000a18a53..1b0f54a13 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher.go
@@ -45,19 +45,17 @@ import (
 
 // Define A Dispatcher Config Struct To Hold Configuration
 type DispatcherConfig struct {
-	Logger             *zap.Logger
-	ClientId           string
-	Brokers            []string
-	Topic              string
-	Username           string
-	Password           string
-	ChannelKey         string
-	StatsReporter      metrics.StatsReporter
-	MetricsRegistry    gometrics.Registry
-	MetricsStopChan    chan struct{}
-	MetricsStoppedChan chan struct{}
-	SaramaConfig       *sarama.Config
-	SubscriberSpecs    []eventingduck.SubscriberSpec
+	Logger          *zap.Logger
+	ClientId        string
+	Brokers         []string
+	Topic           string
+	Username        string
+	Password        string
+	ChannelKey      string
+	StatsReporter   metrics.StatsReporter
+	MetricsRegistry gometrics.Registry
+	SaramaConfig    *sarama.Config
+	SubscriberSpecs []eventingduck.SubscriberSpec
 }
 
 // Knative Eventing SubscriberSpec Wrapper Enhanced With Sarama ConsumerGroup
@@ -87,6 +85,8 @@ type DispatcherImpl struct {
 	subscribers        map[types.UID]*SubscriberWrapper
 	consumerUpdateLock sync.Mutex
 	messageDispatcher  channel.MessageDispatcher
+	MetricsStopChan    chan struct{}
+	MetricsStoppedChan chan struct{}
 }
 
 // Verify The DispatcherImpl Implements The Dispatcher Interface
@@ -97,9 +97,11 @@ func NewDispatcher(dispatcherConfig DispatcherConfig) Dispatcher {
 
 	// Create The DispatcherImpl With Specified Configuration
 	dispatcher := &DispatcherImpl{
-		DispatcherConfig:  dispatcherConfig,
-		subscribers:       make(map[types.UID]*SubscriberWrapper),
-		messageDispatcher: channel.NewMessageDispatcher(dispatcherConfig.Logger),
+		DispatcherConfig:   dispatcherConfig,
+		subscribers:        make(map[types.UID]*SubscriberWrapper),
+		messageDispatcher:  channel.NewMessageDispatcher(dispatcherConfig.Logger),
+		MetricsStopChan:    make(chan struct{}),
+		MetricsStoppedChan: make(chan struct{}),
 	}
 
 	// Start Observing Metrics
diff --git a/pkg/common/config/eventingkafkaconfig.go b/pkg/common/config/eventingkafkaconfig.go
index c77bc93c8..bbc07dd95 100644
--- a/pkg/common/config/eventingkafkaconfig.go
+++ b/pkg/common/config/eventingkafkaconfig.go
@@ -48,6 +48,13 @@ type EKKafkaTopicConfig struct {
 	DefaultRetentionMillis   int64 `json:"defaultRetentionMillis,omitempty"`
 }
 
+// EKCloudEventConfig contains the values send to the Knative cloudevents' ConfigureConnectionArgs function
+// If they are not provided in the configmap, the DefaultMaxIdleConns and DefaultMaxIdleConnsPerHost constants are used
+type EKCloudEventConfig struct {
+	MaxIdleConns        int `json:"maxIdleConns,omitempty"`
+	MaxIdleConnsPerHost int `json:"maxIdleConnsPerHost,omitempty"`
+}
+
 // EKKafkaConfig contains items relevant to Kafka specifically, and the Sarama logging flag
 type EKKafkaConfig struct {
 	Brokers             string             `json:"brokers,omitempty"`
@@ -58,7 +65,8 @@ type EKKafkaConfig struct {
 
 // EventingKafkaConfig is the main struct that holds the Receiver, Dispatcher, and Kafka sub-items
 type EventingKafkaConfig struct {
-	Receiver   EKReceiverConfig   `json:"receiver,omitempty"`
-	Dispatcher EKDispatcherConfig `json:"dispatcher,omitempty"`
-	Kafka      EKKafkaConfig      `json:"kafka,omitempty"`
+	Receiver    EKReceiverConfig   `json:"receiver,omitempty"`
+	Dispatcher  EKDispatcherConfig `json:"dispatcher,omitempty"`
+	CloudEvents EKCloudEventConfig `json:"cloudevents,omitempty"`
+	Kafka       EKKafkaConfig      `json:"kafka,omitempty"`
 }
diff --git a/pkg/common/constants/constants.go b/pkg/common/constants/constants.go
index 43c741803..9d0d53992 100644
--- a/pkg/common/constants/constants.go
+++ b/pkg/common/constants/constants.go
@@ -34,4 +34,8 @@ const (
 
 	// The name of the keys in the Data section of the eventing-kafka configmap that holds Sarama and Eventing-Kafka configuration YAML
 	SaramaSettingsConfigKey = "sarama"
+
+	// Default values for the cloud events connection arguments, if not overridden
+	DefaultMaxIdleConns        = 1000
+	DefaultMaxIdleConnsPerHost = 100
 )


Test patch:
diff --git a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
index a8b039242..8e89ed9e5 100644
--- a/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
+++ b/pkg/channel/consolidated/dispatcher/dispatcher_it_test.go
@@ -27,6 +27,8 @@ import (
 	"testing"
 	"time"
 
+	"knative.dev/eventing-kafka/pkg/common/constants"
+
 	"github.com/cloudevents/sdk-go/v2/binding"
 	"github.com/cloudevents/sdk-go/v2/binding/transformer"
 	protocolhttp "github.com/cloudevents/sdk-go/v2/protocol/http"
@@ -66,8 +68,8 @@ func TestDispatcher(t *testing.T) {
 
 	// Configure connection arguments - to be done exactly once per process
 	kncloudevents.ConfigureConnectionArgs(&kncloudevents.ConnectionArgs{
-		MaxIdleConns:        utils.DefaultMaxIdleConns,
-		MaxIdleConnsPerHost: utils.DefaultMaxIdleConnsPerHost,
+		MaxIdleConns:        constants.DefaultMaxIdleConns,
+		MaxIdleConnsPerHost: constants.DefaultMaxIdleConnsPerHost,
 	})
 
 	dispatcherArgs := KafkaDispatcherArgs{
diff --git a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
index e3cd5f6a0..6210f8b34 100644
--- a/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
+++ b/pkg/channel/distributed/common/kafka/sarama/sarama_test.go
@@ -346,6 +346,10 @@ func verifyTestEKConfigSettings(t *testing.T, saramaConfig *sarama.Config, event
 	assert.Equal(t, commontesting.OldUsername, saramaConfig.Net.SASL.User)
 	assert.Equal(t, commontesting.OldPassword, saramaConfig.Net.SASL.Password)
 	assert.Equal(t, commontesting.DispatcherReplicas, strconv.Itoa(eventingKafkaConfig.Dispatcher.Replicas))
+
+	// Verify that the default CloudEvent settings are set (i.e. not zero)
+	assert.Equal(t, constants.DefaultMaxIdleConns, eventingKafkaConfig.CloudEvents.MaxIdleConns)
+	assert.Equal(t, constants.DefaultMaxIdleConnsPerHost, eventingKafkaConfig.CloudEvents.MaxIdleConnsPerHost)
 }
 
 func getTestSaramaContext(t *testing.T, saramaConfig string, eventingKafkaConfig string) context.Context {
diff --git a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
index f3a496926..0fe44197e 100644
--- a/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
+++ b/pkg/channel/distributed/dispatcher/dispatcher/dispatcher_test.go
@@ -111,6 +111,9 @@ func TestShutdown(t *testing.T) {
 	assert.True(t, consumerGroup2.Closed)
 	assert.True(t, consumerGroup3.Closed)
 	assert.Len(t, dispatcher.subscribers, 0)
+
+	// Verify that calling Shutdown a second time does not cause a panic
+	dispatcher.Shutdown()
 }
 
 // Test The UpdateSubscriptions() Functionality
@@ -376,6 +379,14 @@ func TestConfigChanged(t *testing.T) {
 
 			// 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)
+			}
 		})
 	}
 }
@@ -483,6 +494,14 @@ func TestSecretChanged(t *testing.T) {
 
 			// 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)
+			}
 		})
 	}
 }
@@ -553,21 +572,23 @@ func TestConfigImpl_ObserveMetrics(t *testing.T) {
 
 	reporter := &statsReporterMock{}
 
-	config := &DispatcherImpl{
+	// Create our own DispatcherImpl instead of using NewDispatcher(), so that we can start the metrics
+	// reporting function with a very small interval
+	dispatcher := &DispatcherImpl{
 		DispatcherConfig: DispatcherConfig{
-			Logger:             logtesting.TestLogger(t).Desugar(),
-			MetricsRegistry:    baseSaramaConfig.MetricRegistry,
-			MetricsStopChan:    make(chan struct{}),
-			MetricsStoppedChan: make(chan struct{}),
-			StatsReporter:      reporter,
+			Logger:          logtesting.TestLogger(t).Desugar(),
+			MetricsRegistry: baseSaramaConfig.MetricRegistry,
+			StatsReporter:   reporter,
 		},
+		MetricsStopChan:    make(chan struct{}),
+		MetricsStoppedChan: make(chan struct{}),
 	}
 
 	// Start the metrics observing loop and verify that the report function was called at least once
-	config.ObserveMetrics(5 * time.Millisecond)
+	dispatcher.ObserveMetrics(5 * time.Millisecond)
 	assert.Eventually(t, reporter.GetReportCalled, time.Second, 5*time.Millisecond)
-	close(config.MetricsStopChan)
-	<-config.MetricsStoppedChan
+	close(dispatcher.MetricsStopChan)
+	<-dispatcher.MetricsStoppedChan
 }
 
 // A mock for the StatsReporter that will provide feedback when the Report function is called