Repository: knative-sandbox/eventing-kafka-broker
Pull request #2359 (2022-06-25 21:23:11)

[Prefactoring] Pass brokerClass to Trigger reconciler instead of hardcoding it
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Pass brokerClass to Trigger reconciler instead of hardcoding it
- Prefactoring for KafkaBroker with dataplane-per-namespace
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: 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**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2359#" title="Author self-approved">aliok</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-broker).

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-broker/blob/main/OWNERS)~~ [aliok]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2359?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2359](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2359?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e7f8459) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/b7f108a34e5808141a5e29f6a4632609899b20df?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b7f108a) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@            Coverage Diff            @@
##               main    #2359   +/-   ##
=========================================
  Coverage     81.77%   81.77%           
  Complexity      700      700           
=========================================
  Files            72       72           
  Lines          2414     2414           
  Branches        211      211           
=========================================
  Hits           1974     1974           
  Misses          319      319           
  Partials        121      121           
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.77% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.


------

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

aliok: /hold
Blocked by https://github.com/knative-sandbox/eventing-kafka-broker/pull/2360
aliok: /unhold
pierDipi: /lgtm

Patch:
diff --git a/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml b/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
index cec9e3b19..1c4283b58 100644
--- a/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
+++ b/control-plane/config/post-install/200-storage-version-migrator-cluster-role.yaml
@@ -58,6 +58,19 @@ rules:
       - "update"
       - "patch"
       - "watch"
+  - apiGroups:
+      - "eventing.knative.dev"
+    resources:
+      - "kafkasinks"
+      - "kafkasinks/finalizers"
+      - "kafkasinks/status"
+    verbs:
+      - "get"
+      - "list"
+      - "create"
+      - "update"
+      - "patch"
+      - "watch"
   - apiGroups:
       - ""
     resources:
diff --git a/control-plane/config/post-install/500-storage-version-migrator.yaml b/control-plane/config/post-install/500-storage-version-migrator.yaml
index a7003b860..26a3b70e0 100644
--- a/control-plane/config/post-install/500-storage-version-migrator.yaml
+++ b/control-plane/config/post-install/500-storage-version-migrator.yaml
@@ -42,3 +42,4 @@ spec:
           args:
             - "kafkasources.sources.knative.dev"
             - "kafkachannels.messaging.knative.dev"
+            - "kafkasinks.eventing.knative.dev"
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 824fd6c28..79c3e73a6 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -76,6 +76,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		EventingClient: eventingclient.Get(ctx),
 		Env:            configs,
 		Flags:          feature.Flags{},
+		BrokerClass:    kafka.BrokerClass,
 	}
 
 	impl := triggerreconciler.NewImpl(ctx, reconciler, func(impl *controller.Impl) controller.Options {
@@ -83,7 +84,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 			FinalizerName:     FinalizerName,
 			AgentName:         ControllerAgentName,
 			SkipStatusUpdates: false,
-			PromoteFilterFunc: filterTriggers(reconciler.BrokerLister),
+			PromoteFilterFunc: filterTriggers(reconciler.BrokerLister, kafka.BrokerClass),
 		}
 	})
 
@@ -106,7 +107,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 
 	triggerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: filterTriggers(reconciler.BrokerLister),
+		FilterFunc: filterTriggers(reconciler.BrokerLister, kafka.BrokerClass),
 		Handler:    controller.HandleAll(impl.Enqueue),
 	})
 
@@ -131,7 +132,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 	return impl
 }
 
-func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool {
+func filterTriggers(lister eventinglisters.BrokerLister, brokerClass string) func(interface{}) bool {
 	return func(obj interface{}) bool {
 		trigger, ok := obj.(*eventing.Trigger)
 		if !ok {
@@ -148,7 +149,7 @@ func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool
 		}
 
 		value, ok := broker.GetAnnotations()[apiseventing.BrokerClassKey]
-		return ok && value == kafka.BrokerClass
+		return ok && value == brokerClass
 	}
 }
 
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index 009517784..882387daf 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -39,7 +39,6 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 )
@@ -58,6 +57,8 @@ type Reconciler struct {
 	Env       *config.Env
 	Flags     feature.Flags
 	FlagsLock sync.RWMutex
+
+	BrokerClass string
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
@@ -95,7 +96,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	}
 
 	// Ignore Triggers that are associated with a Broker we don't own.
-	if isKnativeKafkaBroker, brokerClass := isKnativeKafkaBroker(broker); !isKnativeKafkaBroker {
+	if hasRelevantBroker, brokerClass := r.hasRelevantBrokerClass(broker); !hasRelevantBroker {
 		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
 		return nil
 	}
@@ -340,9 +341,9 @@ func deleteTrigger(egresses []*contract.Egress, index int) []*contract.Egress {
 	return egresses[:len(egresses)-1]
 }
 
-func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
+func (r *Reconciler) hasRelevantBrokerClass(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
-	return brokerClass == kafka.BrokerClass, brokerClass
+	return brokerClass == r.BrokerClass, brokerClass
 }
 
 func deliveryOrderFromString(val string) (contract.DeliveryOrder, error) {


Test patch:
diff --git a/control-plane/pkg/reconciler/trigger/controller_test.go b/control-plane/pkg/reconciler/trigger/controller_test.go
index 7f7cf4741..7999af969 100644
--- a/control-plane/pkg/reconciler/trigger/controller_test.go
+++ b/control-plane/pkg/reconciler/trigger/controller_test.go
@@ -170,7 +170,7 @@ func TestFilterTriggers(t *testing.T) {
 			for _, obj := range tc.brokers {
 				_ = brokerInformer.Informer().GetStore().Add(obj)
 			}
-			filter := filterTriggers(brokerInformer.Lister())
+			filter := filterTriggers(brokerInformer.Lister(), kafka.BrokerClass)
 			pass := filter(tc.trigger)
 			assert.Equal(t, tc.pass, pass)
 		})
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 0adea482e..3d5944bf9 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -48,6 +48,7 @@ import (
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -2769,6 +2770,7 @@ func useTableWithFlags(t *testing.T, table TableTest, env *config.Env, flags fea
 			Resolver:       nil,
 			Env:            env,
 			Flags:          flags,
+			BrokerClass:    kafka.BrokerClass,
 		}
 
 		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))