Repository: knative-sandbox/eventing-kafka
Pull request #831 (2021-08-25 01:50:54)

Use `controller.Impl` tracker and cleanups
- 🧽 Update or clean up current behavior

/kind cleanup

**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 **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/831#" title="Author self-approved">mattmoor</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **zhiminxiang** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @zhiminxiang` 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":["zhiminxiang"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#831](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (51b7ce5) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/2c7cc36ca74a15939437f496842cca2059291154?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2c7cc36) will **increase** coverage by `0.01%`.
> The diff coverage is `33.33%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831/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/831?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     #831      +/-   ##
==========================================
+ Coverage   75.29%   75.30%   +0.01%     
==========================================
  Files         139      139              
  Lines        6440     6439       -1     
==========================================
  Hits         4849     4849              
+ Misses       1363     1362       -1     
  Partials      228      228              
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [pkg/source/reconciler/source/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL3NvdXJjZS9yZWNvbmNpbGVyL3NvdXJjZS9jb250cm9sbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [.../distributed/dispatcher/controller/kafkachannel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-cGtnL2NoYW5uZWwvZGlzdHJpYnV0ZWQvZGlzcGF0Y2hlci9jb250cm9sbGVyL2thZmthY2hhbm5lbC5nbw==) | `84.87% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831?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/831?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [2c7cc36...51b7ce5](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/831?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 5c593bd48..a6f9fa926 100644
--- a/cmd/channel/distributed/dispatcher/main.go
+++ b/cmd/channel/distributed/dispatcher/main.go
@@ -162,6 +162,7 @@ func main() {
 
 	// Construct The KafkaChannel Controller
 	kcController := controller.NewController(
+		ctx,
 		logger,
 		environment.ChannelKey,
 		dispatcher,
diff --git a/go.mod b/go.mod
index c20e45680..c3c4eb7bf 100644
--- a/go.mod
+++ b/go.mod
@@ -35,9 +35,9 @@ require (
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
 	knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b
-	knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57
+	knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
-	knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08
-	knative.dev/pkg v0.0.0-20210823133635-12d915ede293
+	knative.dev/networking v0.0.0-20210824140523-51512a042e23
+	knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3
 	knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087
 )
diff --git a/go.sum b/go.sum
index 7fa7f9f8c..c8b7f91fd 100644
--- a/go.sum
+++ b/go.sum
@@ -1274,18 +1274,17 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b h1:1kfFg/0CbbaVyNlT0JCYQ/6IctbUibaDiNFJRUShnKA=
 knative.dev/control-protocol v0.0.0-20210823140136-471903b6c44b/go.mod h1:9gYmAq6zVKUA+p1+DXslH91jtga5q2z+YJWcH+z88EE=
-knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57 h1:dIZgOeO7vWeCFBVE2vM9UCseOmw8YaKTW4rzizO4viU=
-knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57/go.mod h1:qGd3YZqi++xwk0WNaX0VYMjpywq2JWptL6l+uR+Jhh4=
+knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce h1:OHUkMUIFbsmOjINnoB0+q4GqdmQ79k1VHbHbGH9toek=
+knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce/go.mod h1:J6C3diGWNurw71eAPLuVeILNR4SSqu+YSMePyovq0rY=
 knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c h1:nOXoDWAAItwr4o0dp3nHr6skgpVD4IvME/UX84YNl5k=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210806075220-815cd312d65c/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08 h1:paj2ByFR/7l0WuqCRHlbr2vl1QUc9qXn2+L2G0lkZXk=
-knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08/go.mod h1:4lsqjd2G5aHfh0zJcA+G9F2gh0r0XSYn9vEygCLXDd4=
+knative.dev/networking v0.0.0-20210824140523-51512a042e23 h1:1V//6gQw95MecpJFICaDZwIYlK8gFflXMjt9ZNhnIss=
+knative.dev/networking v0.0.0-20210824140523-51512a042e23/go.mod h1:UmOQHaS6K3xngvhW5SyEOFy2nFltL9BtvroDjysYXSw=
 knative.dev/pkg v0.0.0-20210803160015-21eb4c167cc5/go.mod h1:RPk5txNA3apR9X40D4MpUOP9/VqOG8CrtABWfOwGVS4=
-knative.dev/pkg v0.0.0-20210822203034-3bed4dcf179c/go.mod h1:t6WzDPywQloi+AbOtiOejfpem1mLfKeuV+sjXGQLGN8=
-knative.dev/pkg v0.0.0-20210823133635-12d915ede293 h1:6GvpU4ESZxfb+trIhvMTZcZSfGGeIf+F1aG7Fv3R9gA=
-knative.dev/pkg v0.0.0-20210823133635-12d915ede293/go.mod h1:67yUYAgAWesv4ehCShUuIxkaxrsROSLQr3kApW4xPO4=
+knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3 h1:38SNaSkVBI1Fkc4Lk2AMEjPW03h3cAwTS3+dKK9RV+M=
+knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3/go.mod h1:67yUYAgAWesv4ehCShUuIxkaxrsROSLQr3kApW4xPO4=
 knative.dev/reconciler-test v0.0.0-20210803183715-b61cc77c06f6/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
 knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087 h1:HVcaI8P+CNHl5VgnqWgAzGoc9vbvf3D421rd2qEo3Wc=
 knative.dev/reconciler-test v0.0.0-20210820180205-a25de6a08087/go.mod h1:+Kovy+G5zXZNcuO/uB+zfo37vFKZzsLIlWezt/nKMz0=
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
index c8be54863..09f6ef885 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel.go
@@ -72,6 +72,7 @@ var _ controller.Reconciler = Reconciler{}
 // NewController initializes the controller and is called by the generated code.
 // Registers event handlers to enqueue events.
 func NewController(
+	ctx context.Context,
 	logger *zap.Logger,
 	channelKey string,
 	dispatcher dispatcher.Dispatcher,
@@ -90,7 +91,7 @@ func NewController(
 		kafkachannelLister:   kafkachannelInformer.Lister(),
 		kafkaClientSet:       kafkaClientSet,
 	}
-	reconciler.impl = controller.NewImplFull(reconciler, controller.ControllerOptions{
+	reconciler.impl = controller.NewContext(ctx, reconciler, controller.ControllerOptions{
 		WorkQueueName: ReconcilerName,
 		Logger:        logger.Sugar(),
 	})
diff --git a/pkg/source/reconciler/binding/controller.go b/pkg/source/reconciler/binding/controller.go
index 5ce75ffdf..330bb4f88 100644
--- a/pkg/source/reconciler/binding/controller.go
+++ b/pkg/source/reconciler/binding/controller.go
@@ -36,7 +36,6 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/clients/dynamicclient"
 	"knative.dev/pkg/logging"
-	"knative.dev/pkg/tracker"
 	"knative.dev/pkg/webhook/psbinding"
 )
 
@@ -81,14 +80,15 @@ func NewController(
 			scheme.Scheme, corev1.EventSource{Component: controllerAgentName}),
 		NamespaceLister: namespaceInformer.Lister(),
 	}
-	impl := controller.NewImpl(c, logger, "KafkaBindings")
-
-	logger.Info("Setting up event handlers")
+	impl := controller.NewContext(ctx, c, controller.ControllerOptions{
+		Logger:        logger,
+		WorkQueueName: "KafkaBindings",
+	})
 
 	kfkInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 	namespaceInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
-	c.Tracker = tracker.New(impl.EnqueueKey, controller.GetTrackerLease(ctx))
+	c.Tracker = impl.Tracker
 	c.Factory = &duck.CachedInformerFactory{
 		Delegate: &duck.EnqueueInformerFactory{
 			Delegate:     psInformerFactory,
diff --git a/pkg/source/reconciler/mtsource/controller.go b/pkg/source/reconciler/mtsource/controller.go
index 68f170565..79dfbdc5c 100644
--- a/pkg/source/reconciler/mtsource/controller.go
+++ b/pkg/source/reconciler/mtsource/controller.go
@@ -75,7 +75,7 @@ func NewController(
 
 	impl := kafkasource.NewImpl(ctx, c)
 
-	c.sinkResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+	c.sinkResolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	// Use a different set of conditions
 	sourcesv1beta1.RegisterAlternateKafkaConditionSet(sourcesv1beta1.KafkaMTSourceCondSet)
diff --git a/pkg/source/reconciler/source/controller.go b/pkg/source/reconciler/source/controller.go
index c28fa50c3..a06f8e52c 100644
--- a/pkg/source/reconciler/source/controller.go
+++ b/pkg/source/reconciler/source/controller.go
@@ -67,16 +67,14 @@ func NewController(
 	}
 
 	impl := kafkasource.NewImpl(ctx, c)
-	c.sinkResolver = resolver.NewURIResolver(ctx, impl.EnqueueKey)
+	c.sinkResolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	c.claimsNotificationStore = ctrlreconciler.NewNotificationStore(impl.EnqueueKey, kafkasourcecontrol.ClaimsParser)
 
-	logging.FromContext(ctx).Info("Setting up kafka event handlers")
-
 	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
 
 	deploymentInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterControllerGK(v1beta1.Kind("KafkaSource")),
+		FilterFunc: controller.FilterController(&v1beta1.KafkaSource{}),
 		Handler:    controller.HandleAll(impl.EnqueueControllerOf),
 	})
 
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
index 1d77996cf..3aa377a52 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_lifecycle.go
@@ -73,9 +73,12 @@ func (imcs *InMemoryChannelStatus) GetCondition(t apis.ConditionType) *apis.Cond
 	return imcCondSet.Manage(imcs).GetCondition(t)
 }
 
-// IsReady returns true if the resource is ready overall.
-func (imcs *InMemoryChannelStatus) IsReady() bool {
-	return imcCondSet.Manage(imcs).IsHappy()
+// IsReady returns true if the Status condition InMemoryChannelConditionReady
+// is true and the latest spec has been observed.
+func (imc *InMemoryChannel) IsReady() bool {
+	imcs := imc.Status
+	return imcs.ObservedGeneration == imc.Generation &&
+		imcCondSet.Manage(&imcs).IsHappy()
 }
 
 // InitializeConditions sets relevant unset conditions to Unknown state.
diff --git a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
index ca295cb52..5b0c971b5 100644
--- a/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/messaging/v1/in_memory_channel_types.go
@@ -71,7 +71,7 @@ type InMemoryChannelSpec struct {
 
 // ChannelStatus represents the current state of a Channel.
 type InMemoryChannelStatus struct {
-	// Channel conforms to Duck type Channelable.
+	// Channel conforms to Duck type ChannelableStatus.
 	eventingduckv1.ChannelableStatus `json:",inline"`
 }
 
diff --git a/vendor/knative.dev/pkg/configmap/parse.go b/vendor/knative.dev/pkg/configmap/parse.go
index 4c280757b..faeacab14 100644
--- a/vendor/knative.dev/pkg/configmap/parse.go
+++ b/vendor/knative.dev/pkg/configmap/parse.go
@@ -141,7 +141,11 @@ func AsDuration(key string, target *time.Duration) ParseFunc {
 func AsStringSet(key string, target *sets.String) ParseFunc {
 	return func(data map[string]string) error {
 		if raw, ok := data[key]; ok {
-			*target = sets.NewString(strings.Split(raw, ",")...)
+			splitted := strings.Split(raw, ",")
+			for i, v := range splitted {
+				splitted[i] = strings.TrimSpace(v)
+			}
+			*target = sets.NewString(splitted...)
 		}
 		return nil
 	}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 07c57633b..ac5710782 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1037,7 +1037,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.25.1-0.20210823153835-b2700c2dcf57
+# knative.dev/eventing v0.25.1-0.20210824204225-4f6f3075d2ce
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1164,10 +1164,10 @@ knative.dev/eventing/test/upgrade/prober/wathola/sender
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/networking v0.0.0-20210823062535-cfb7e2379d08
+# knative.dev/networking v0.0.0-20210824140523-51512a042e23
 ## explicit
 knative.dev/networking/pkg/prober
-# knative.dev/pkg v0.0.0-20210823133635-12d915ede293
+# knative.dev/pkg v0.0.0-20210824120823-a94f5f07b3c3
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate


Test patch:
diff --git a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
index 64c37bc65..cb2583e9a 100644
--- a/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
+++ b/pkg/channel/distributed/dispatcher/controller/kafkachannel_test.go
@@ -88,7 +88,7 @@ func TestNewController(t *testing.T) {
 			stopChan := make(chan struct{})
 
 			// Perform The Test
-			c := NewController(logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan, testCase.managerEvents)
+			c := NewController(context.TODO(), logger, channelKey, mockDispatcher, kafkaChannelInformer, fakeK8sClientSet, fakeKafkaChannelClientSet, stopChan, testCase.managerEvents)
 
 			// Verify Results
 			assert.NotNil(t, c)
@@ -122,7 +122,7 @@ func TestProcessManagerEvents(t *testing.T) {
 		kafkachannelInformer: kafkaChannelInformer.Informer(),
 		kafkachannelLister:   kafkaChannelInformer.Lister(),
 		kafkaClientSet:       fakeKafkaChannelClientSet,
-		impl:                 controller.NewImplFull(mockReconciler, controller.ControllerOptions{Logger: logger.Sugar()}),
+		impl:                 controller.NewContext(context.TODO(), mockReconciler, controller.ControllerOptions{Logger: logger.Sugar()}),
 	}
 	events := make(chan consumer.ManagerEvent)