Repository: knative-sandbox/eventing-kafka-broker
Pull request #1610 (2021-12-14 20:02:04)

[data-plane]: Bump debezium-core from 1.7.1.Final to 1.7.2.Final in /data-plane
Bumps debezium-core from 1.7.1.Final to 1.7.2.Final.


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=io.debezium:debezium-core&package-manager=maven&previous-version=1.7.1.Final&new-version=1.7.2.Final)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

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/1610#" 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 **matzew** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @matzew` 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":["matzew"]} -->
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
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1610?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1610](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1610?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (5adb763) 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 **not change** coverage.
> The diff coverage is `n/a`.

> :exclamation: Current head 5adb763 differs from pull request most recent head 22def9d. Consider uploading reports for the commit 22def9d to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1610/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/1610?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    #1610   +/-   ##
=========================================
  Coverage     73.07%   73.07%           
- Complexity      595      596    +1     
=========================================
  Files           118      118           
  Lines          4375     4375           
  Branches        171      171           
=========================================
  Hits           3197     3197           
  Misses          914      914           
  Partials        264      264           
```

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


------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1610?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/1610?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [129af86...22def9d](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1610?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: @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-integration-tests | 22def9d532a6799bbb02765d675031b1973f9149 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1610/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1470849769113915392) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1610). [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 -->

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/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/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 5d87484fb..9936c8948 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"
 
@@ -88,16 +86,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)
+			}
+		})
+	}
+}