Repository: knative-sandbox/eventing-kafka-broker
Pull request #1609 (2021-12-14 20:01:53)

[data-plane]: Bump ch.qos.logback.version from 1.2.7 to 1.2.8 in /data-plane
Bumps `ch.qos.logback.version` from 1.2.7 to 1.2.8.
Updates `logback-core` from 1.2.7 to 1.2.8

Updates `logback-classic` from 1.2.7 to 1.2.8


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

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-broker/pull/1609#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **pierdipi** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @pierdipi` 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-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/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":["pierdipi"]} -->
knative-prow-robot: Hi @dependabot[bot]. 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-broker).

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

pierDipi: /ok-to-test
knative-prow-robot: @dependabot[bot]: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-build-tests | 4cee5f20057541b3c07f5abfe3bdce316baea0c5 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1609/pull-knative-sandbox-eventing-kafka-broker-build-tests/1470846891922034688) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1609). [Your PR dashboard](https://gubernator.knative.dev/pr/dependabot[bot]).

<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. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1609?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1609](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1609?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4cee5f2) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/129af86ede61cd0f93aaab2e8c1b612d16b84e59?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (129af86) will **decrease** coverage by `0.11%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1609/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/1609?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    #1609      +/-   ##
============================================
- Coverage     73.07%   72.96%   -0.12%     
  Complexity      595      595              
============================================
  Files           118      118              
  Lines          4375     4375              
  Branches        171      171              
============================================
- Hits           3197     3192       -5     
- Misses          914      918       +4     
- Partials        264      265       +1     
```

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

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/1609?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/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1609/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `60.00% <0.00%> (-15.00%)` | :arrow_down: |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1609/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `95.23% <0.00%> (-4.77%)` | :arrow_down: |

------

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

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-broker/pull/1609#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please ask for approval from **pierdipi** 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-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/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":["pierdipi"]} -->

Patch:
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 272a9cd36..33645cf7a 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -20,6 +20,9 @@ import (
 	"context"
 	"log"
 
+	"knative.dev/pkg/injection"
+	"knative.dev/pkg/signals"
+
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/sharedmain"
@@ -52,27 +55,38 @@ func main() {
 		log.Fatal("cannot process environment variables with prefix SINK", err)
 	}
 
-	sharedmain.Main(
-		component,
+	sharedmain.MainNamed(signals.NewContext(), component,
 
 		// Broker controller
-		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return broker.NewController(ctx, watcher, brokerEnv)
+		injection.NamedControllerConstructor{
+			Name: "broker-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return broker.NewController(ctx, watcher, brokerEnv)
+			},
 		},
 
 		// Trigger controller
-		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return trigger.NewController(ctx, watcher, brokerEnv)
+		injection.NamedControllerConstructor{
+			Name: "trigger-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return trigger.NewController(ctx, watcher, brokerEnv)
+			},
 		},
 
 		// KafkaSink controller
-		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return sink.NewController(ctx, watcher, sinkEnv)
+		injection.NamedControllerConstructor{
+			Name: "sink-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return sink.NewController(ctx, watcher, sinkEnv)
+			},
 		},
 
 		// KafkaSource controller
-		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return source.NewController(ctx, watcher, sourceEnv)
+		injection.NamedControllerConstructor{
+			Name: "source-controller",
+			ControllerConstructor: func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
+				return source.NewController(ctx, watcher, sourceEnv)
+			},
 		},
 	)
 }
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..cc5cbd15b 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,7 +1,7 @@
 
 Lists of 158 third-party dependencies.
-     (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.7 - http://logback.qos.ch/logback-classic)
-     (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.7 - http://logback.qos.ch/logback-core)
+     (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.8 - http://logback.qos.ch/logback-classic)
+     (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.8 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.11.4 - http://github.com/FasterXML/jackson)
      (The Apache Software License, Version 2.0) Jackson-core (com.fasterxml.jackson.core:jackson-core:2.11.4 - https://github.com/FasterXML/jackson-core)
      (The Apache Software License, Version 2.0) jackson-databind (com.fasterxml.jackson.core:jackson-databind:2.11.4 - http://github.com/FasterXML/jackson)
@@ -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..23c694a16 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -48,7 +48,7 @@
     <jackson.version>2.13.0</jackson.version>
     <protobuf.version>3.19.1</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
-    <ch.qos.logback.version>1.2.7</ch.qos.logback.version>
+    <ch.qos.logback.version>1.2.8</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>
     <assertj.version>3.21.0</assertj.version>
     <awaitility.version>4.1.1</awaitility.version>
@@ -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)
+			}
+		})
+	}
+}