Repository: knative-sandbox/eventing-kafka-broker
Pull request #1708 (2022-01-07 15:50:27)

Enqueue trigger when ConsumerGroups change
We need to enqueue the Trigger key associated with the 
ConsumerGroup and not the ConsumerGroup resource.

Related https://github.com/knative-sandbox/eventing-kafka-broker/pull/1611#discussion_r780340587

## Proposed Changes

- Enqueue trigger when ConsumerGroups change

<!--
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.
-->

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/1708#" 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":[]} -->
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/consumergroup/filter.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1708/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1479480564296716288/artifacts/line-cov.html#file0) | 100.0% | 52.9% | -47.1

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1708](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ac31819) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/fad14667c1b72c7bf6487ee20698ee186fa3bc40?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (fad1466) will **decrease** coverage by `7.00%`.
> The diff coverage is `12.50%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/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/1708?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    #1708      +/-   ##
============================================
- Coverage     72.65%   65.64%   -7.01%     
============================================
  Files           120       54      -66     
  Lines          4443     2521    -1922     
  Branches        171        0     -171     
============================================
- Hits           3228     1655    -1573     
+ Misses          951      693     -258     
+ Partials        264      173      -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/1708?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | `53.33% <0.00%> (-46.67%)` | :arrow_down: |
| [...ol-plane/pkg/reconciler/trigger/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | `58.49% <100.00%> (ø)` | |
| [...venting/kafka/broker/dispatcher/DeliveryOrder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvRGVsaXZlcnlPcmRlci5qYXZh) | | |
| [...ka/broker/dispatcher/impl/filter/PrefixFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvUHJlZml4RmlsdGVyLmphdmE=) | | |
| [...er/dispatcher/impl/consumer/InvalidCloudEvent.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9JbnZhbGlkQ2xvdWRFdmVudC5qYXZh) | | |
| [...fka/broker/receiver/impl/handler/ProbeHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvUHJvYmVIYW5kbGVyLmphdmE=) | | |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [...enting/kafka/broker/receiver/main/ReceiverEnv.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyRW52LmphdmE=) | | |
| [...ive/eventing/kafka/broker/core/utils/Shutdown.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvU2h1dGRvd24uamF2YQ==) | | |
| [...fka/broker/dispatcher/impl/filter/ExactFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvRXhhY3RGaWx0ZXIuamF2YQ==) | | |
| ... and [60 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708/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/1708?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/1708?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [fad1466...ac31819](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1708?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 **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1708#pullrequestreview-847009359" title="LGTM">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1708#" 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":[]} -->

Patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/filter.go b/control-plane/pkg/reconciler/consumergroup/filter.go
index 1ce2b2a2f..0ff1b0927 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter.go
@@ -19,6 +19,8 @@ package consumergroup
 import (
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
@@ -41,3 +43,20 @@ func Filter(userFacingResource string) func(obj interface{}) bool {
 		return false
 	}
 }
+
+// Enqueue enqueues using the provided enqueue function the resource associated with a ConsumerGroup
+func Enqueue(userFacingResource string, enqueue func(key types.NamespacedName)) func(obj interface{}) {
+	userFacingResource = strings.ToLower(userFacingResource)
+	return func(obj interface{}) {
+		cg, ok := obj.(*kafkainternals.ConsumerGroup)
+		if !ok {
+			return
+		}
+
+		for _, or := range cg.OwnerReferences {
+			if strings.ToLower(or.Kind) == userFacingResource {
+				enqueue(types.NamespacedName{Namespace: cg.GetNamespace(), Name: or.Name})
+			}
+		}
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 8f0ec27df..22229f0b9 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -88,7 +88,7 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 	// ConsumerGroup changes and enqueue associated Trigger
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: consumergroup.Filter("trigger"),
-		Handler:    controller.HandleAll(impl.Enqueue),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("trigger", impl.EnqueueKey)),
 	})
 
 	return impl


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/filter_test.go b/control-plane/pkg/reconciler/consumergroup/filter_test.go
index e56dc59b3..5f00b23c7 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
@@ -76,3 +77,61 @@ func TestFilter(t *testing.T) {
 		})
 	}
 }
+
+func TestEnqueue(t *testing.T) {
+	tests := []struct {
+		name               string
+		resource           interface{}
+		userFacingResource string
+		want               int
+	}{
+		{
+			name:               "unknown type",
+			resource:           &kafkainternals.Consumer{},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+		{
+			name: "trigger pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "trigger",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               1,
+		},
+		{
+			name: "no pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "kafkasource",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			called := 0
+			f := func(key types.NamespacedName) {
+				called++
+			}
+			Enqueue(tt.userFacingResource, f)(tt.resource)
+			if called != tt.want {
+				t.Errorf("Enqueue called %v, want %v", called, tt.want)
+			}
+		})
+	}
+}