Repository: knative-sandbox/eventing-kafka-broker
Pull request #1616 (2021-12-15 13:37:00)

Add promote filter function to Broker and Trigger reconciler
When a reconciler gets promoted to be the leader, every object
gets re-queued even if it shouldn't be.

Tested in #1612

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Add promote filter function to Broker and Trigger reconciler

<!--
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
None
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

```
None
```

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

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

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":[]} -->
pierDipi: /cherry-pick release-1.1
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.1 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1616#issuecomment-994799477):

>/cherry-pick release-1.1


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>
pierDipi: /cherry-pick release-1.0
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.0 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1616#issuecomment-994799528):

>/cherry-pick release-1.0


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>
pierDipi: /cherry-pick release-0.26
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-0.26 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/1616#issuecomment-994799600):

>/cherry-pick release-0.26


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-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1616/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1471112058119393280/artifacts/line-cov.html#file0) | 85.2% | 85.7% | 0.5

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1616](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6da1be0) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/2e8d0112367fb6a4e62336e0ba102ed3b3c4f72b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2e8d011) will **decrease** coverage by `6.87%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&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-broker/pull/1616?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    #1616      +/-   ##
============================================
- Coverage     73.12%   66.24%   -6.88%     
============================================
  Files           118       52      -66     
  Lines          4375     2450    -1925     
  Branches        171        0     -171     
============================================
- Hits           3199     1623    -1576     
+ Misses          911      653     -258     
+ Partials        265      174      -91     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | `79.68% <100.00%> (+0.65%)` | :arrow_up: |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | `85.29% <100.00%> (+0.21%)` | :arrow_up: |
| [...ol-plane/pkg/reconciler/trigger/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | `50.90% <100.00%> (+0.90%)` | :arrow_up: |
| [...nting/kafka/broker/core/security/AuthProvider.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvQXV0aFByb3ZpZGVyLmphdmE=) | | |
| [...kafka/broker/dispatcher/impl/filter/SqlFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvU3FsRmlsdGVyLmphdmE=) | | |
| [...spatcher/impl/consumer/CloudEventDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50RGVzZXJpYWxpemVyLmphdmE=) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| [...er/impl/consumer/InvalidCloudEventInterceptor.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudEludGVyY2VwdG9yLmphdmE=) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...e/eventing/kafka/broker/core/file/FileWatcher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZmlsZS9GaWxlV2F0Y2hlci5qYXZh) | | |
| ... and [59 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616?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/1616?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [2e8d011...6da1be0](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1616?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/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 798d178b2..d488bee56 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -83,7 +83,9 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 		)
 	}
 
-	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
+	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass, func(impl *controller.Impl) controller.Options {
+		return controller.Options{PromoteFilterFunc: kafka.BrokerClassFilter()}
+	})
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
 	reconciler.Prober = prober.NewAsync(ctx, http.DefaultClient, env.IngressPodPort, reconciler.ReceiverSelector(), impl.EnqueueKey)
diff --git a/control-plane/pkg/reconciler/consumergroup/filter.go b/control-plane/pkg/reconciler/consumergroup/filter.go
new file mode 100644
index 000000000..1ce2b2a2f
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/filter.go
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package consumergroup
+
+import (
+	"strings"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+)
+
+// Filter returns a filter function based on the user-facing resource that a controller is tracking.
+// Usable by FilteringResourceEventHandler.
+func Filter(userFacingResource string) func(obj interface{}) bool {
+	userFacingResource = strings.ToLower(userFacingResource)
+	return func(obj interface{}) bool {
+		cg, ok := obj.(*kafkainternals.ConsumerGroup)
+		if !ok {
+			return false
+		}
+
+		for _, or := range cg.OwnerReferences {
+			if strings.ToLower(or.Kind) == userFacingResource {
+				return true
+			}
+		}
+
+		return false
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 8f4bffe99..cee3304f8 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -81,6 +81,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 			FinalizerName:     FinalizerName,
 			AgentName:         ControllerAgentName,
 			SkipStatusUpdates: false,
+			PromoteFilterFunc: filterTriggers(reconciler.BrokerLister),
 		}
 	})
 
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 5d87484fb..8f0ec27df 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -18,11 +18,9 @@ package v2
 
 import (
 	"context"
-	"strings"
 
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/labels"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
@@ -33,9 +31,9 @@ import (
 	triggerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/trigger"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 
-	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
 	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
 
@@ -72,6 +70,7 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 			FinalizerName:     FinalizerName,
 			AgentName:         ControllerAgentName,
 			SkipStatusUpdates: false,
+			PromoteFilterFunc: filterTriggers(reconciler.BrokerLister),
 		}
 	})
 
@@ -88,16 +87,10 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 
 	// ConsumerGroup changes and enqueue associated Trigger
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		Handler: controller.HandleAll(func(obj interface{}) {
-			if cg, ok := obj.(*internalscg.ConsumerGroup); ok {
-				for _, or := range cg.OwnerReferences {
-					if strings.ToLower(or.Kind) == "trigger" {
-						impl.EnqueueKey(types.NamespacedName{Namespace: cg.GetNamespace(), Name: or.Name}) //enqueue trigger with name from owner ref
-						return
-					}
-				}
-			}
-		})})
+		FilterFunc: consumergroup.Filter("trigger"),
+		Handler:    controller.HandleAll(impl.Enqueue),
+	})
+
 	return impl
 }
 
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index 871e628b5..89d01e6a5 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -41,8 +41,8 @@ Lists of 158 third-party dependencies.
      (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
      (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (The Apache Software License, Version 2.0) cloudevents-sql (io.cloudevents:cloudevents-sql:2.2.0 - https://cloudevents.github.io/sdk-java/cloudevents-sql/)
-     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.7.1.Final - https://debezium.io/debezium-parent/debezium-api)
-     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.7.1.Final - https://debezium.io/debezium-parent/debezium-core)
+     (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.7.2.Final - https://debezium.io/debezium-parent/debezium-api)
+     (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.7.2.Final - https://debezium.io/debezium-parent/debezium-core)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes :: Java Client (io.fabric8:kubernetes-client:5.10.1 - http://fabric8.io/kubernetes-client/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: Admission Registration, Authentication and Authorization (io.fabric8:kubernetes-model-admissionregistration:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-admissionregistration/)
      (Apache License, Version 2.0) Fabric8 :: Kubernetes Model :: API Extensions (io.fabric8:kubernetes-model-apiextensions:5.10.1 - http://fabric8.io/kubernetes-model-generator/kubernetes-model-apiextensions/)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index dc6ea2968..78105f2c3 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -56,7 +56,7 @@
     <mokito.junit.jupiter.version>4.1.0</mokito.junit.jupiter.version>
     <fabric8.kubernetes.version>5.10.1</fabric8.kubernetes.version>
     <kafka.version>2.8.1</kafka.version>
-    <debezium.version>1.7.1.Final</debezium.version>
+    <debezium.version>1.7.2.Final</debezium.version>
   </properties>
 
   <modules>


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/filter_test.go b/control-plane/pkg/reconciler/consumergroup/filter_test.go
new file mode 100644
index 000000000..e56dc59b3
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumergroup/filter_test.go
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package consumergroup
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+)
+
+func TestFilter(t *testing.T) {
+	tests := []struct {
+		name               string
+		resource           interface{}
+		userFacingResource string
+		want               bool
+	}{
+		{
+			name:               "unknown type",
+			resource:           &kafkainternals.Consumer{},
+			userFacingResource: "trigger",
+			want:               false,
+		},
+		{
+			name: "trigger pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "trigger",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               true,
+		},
+		{
+			name: "no pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "kafkasource",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := Filter(tt.userFacingResource)(tt.resource); got != tt.want {
+				t.Errorf("Filter() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}