Repository: knative-sandbox/eventing-kafka
Pull request #493 (2021-03-31 18:05:26)

Move metrics shutdown channels out of dispatcher config
## Proposed Changes

-🐛 The introduction of the Sarama metrics to the dispatcher (https://github.com/knative-sandbox/eventing-kafka/pull/464) included two channels (MetricsStopChan and MetricsStoppedChan) that were inappropriately added to the DispatcherConfig instead of the DispatcherImpl.  Since the config is re-used between creation of dispatchers, these closed channels were copied to the new DispatcherImpl's config and caused a panic when Shutdown is called again.  Moving these channels to the DispatcherImpl, the fields of which are recreated each time, solves this issue.


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 **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-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>

travis-minke-sap: /ok-to-test
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/493?src=pr&el=h1) Report
> Merging [#493](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/493?src=pr&el=desc) (c0f8268) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/b0bad2589f0ca823013fa45d84a6aa2dd8fdfb96?el=desc) (b0bad25) will **increase** coverage by `0.06%`.
> The diff coverage is `100.00%`.

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

```diff
@@            Coverage Diff             @@
##             main     #493      +/-   ##
==========================================
+ Coverage   73.82%   73.88%   +0.06%     
==========================================
  Files         139      139              
  Lines        5601     5603       +2     
==========================================
+ Hits         4135     4140       +5     
+ Misses       1219     1217       -2     
+ Partials      247      246       -1     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/493?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [...el/distributed/dispatcher/dispatcher/dispatcher.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/493/diff?src=pr&el=tree#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9kaXNwYXRjaGVyL2Rpc3BhdGNoZXIuZ28=) | `79.28% <100.00%> (+2.47%)` | :arrow_up: |

------

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

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/493/pull-knative-sandbox-eventing-kafka-go-coverage/1377322255897333760/artifacts/line-cov.html#file0) | 80.2% | 81.7% | 1.6

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/493/pull-knative-sandbox-eventing-kafka-go-coverage/1377324429515689984/artifacts/line-cov.html#file0) | 80.2% | 81.7% | 1.6

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/493/pull-knative-sandbox-eventing-kafka-go-coverage/1377336052577996800/artifacts/line-cov.html#file0) | 80.2% | 86.5% | 6.3

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/493#pullrequestreview-625743306" title="Approved">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 2519846f7..331604adb 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -135,18 +135,16 @@ func main() {
 
 	// 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/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


Test patch:
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