Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Deleting Broker before Trigger leaves the Trigger's kafka.triggers.eventing.knative.dev finalizer stuck
**Describe the bug**
Deleting a Kafka-classed Broker first, and then its Trigger second makes the trigger deletion stuck, with the kafka.triggers.eventing.knative.dev finalizer blocking the trigger deletion

 **Expected behavior**
It should be possible to delete the Broker first, and then its Triggers. 

 **To Reproduce**
1. Create a Kafka Broker with Triggers, e.g https://gist.github.com/maschmid/360ceff1e0e1f9b45eafb87a7425a75f 
2. Delete the Broker
3. Delete one of the triggers
4. Notice the trigger deletion is stuck

 **Knative release version**
0.26

 **Additional context**
installed via https://github.com/openshift-knative/serverless-operator (current main, 1.20)


Pull request description:
[release-0.26] Remove Trigger finalizer when Broker is removed
Fixes #1590

This is an automated cherry-pick of #1350

```release-note
None
```

Patch:
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index e44306156..425253dee 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -119,6 +119,10 @@ func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool
 			return false
 		}
 
+		if hasKafkaBrokerTriggerFinalizer(trigger.Finalizers, FinalizerName) {
+			return true
+		}
+
 		broker, err := lister.Brokers(trigger.Namespace).Get(trigger.Spec.Broker)
 		if err != nil {
 			return false
@@ -129,6 +133,15 @@ func filterTriggers(lister eventinglisters.BrokerLister) func(interface{}) bool
 	}
 }
 
+func hasKafkaBrokerTriggerFinalizer(finalizers []string, finalizerName string) bool {
+	for _, f := range finalizers {
+		if f == finalizerName {
+			return true
+		}
+	}
+	return false
+}
+
 func enqueueTriggers(
 	logger *zap.Logger,
 	lister eventinglisters.TriggerLister,


Test patch:
diff --git a/control-plane/pkg/reconciler/trigger/controller_test.go b/control-plane/pkg/reconciler/trigger/controller_test.go
index 4a4d2cbe5..e254cd28c 100644
--- a/control-plane/pkg/reconciler/trigger/controller_test.go
+++ b/control-plane/pkg/reconciler/trigger/controller_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
 	"knative.dev/pkg/configmap"
@@ -34,6 +35,7 @@ import (
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 func TestNewController(t *testing.T) {
@@ -48,11 +50,124 @@ func TestNewController(t *testing.T) {
 func TestFilterTriggers(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	pass := filterTriggers(brokerinformer.Get(ctx).Lister())(&eventing.Trigger{
-		Spec: eventing.TriggerSpec{
-			Broker: "not-exists",
+	tt := []struct {
+		name    string
+		trigger interface{}
+		pass    bool
+		brokers []*eventing.Broker
+	}{
+		{
+			name:    "unknown type",
+			trigger: &eventing.Broker{},
+			pass:    false,
 		},
-	})
+		{
+			name: "non existing broker",
+			trigger: &eventing.Trigger{
+				Spec: eventing.TriggerSpec{
+					Broker: "not-exists",
+				},
+			},
+			pass: false,
+		},
+		{
+			name: "non existing broker and trigger with kafka broker finalizer",
+			trigger: &eventing.Trigger{
+				ObjectMeta: metav1.ObjectMeta{
+					Finalizers: []string{FinalizerName},
+				},
+				Spec: eventing.TriggerSpec{
+					Broker: "not-exists",
+				},
+			},
+			pass: true,
+		},
+		{
+			name: "exiting kafka broker",
+			trigger: &eventing.Trigger{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace:  "ns",
+					Name:       "tr",
+					Finalizers: []string{FinalizerName},
+				},
+				Spec: eventing.TriggerSpec{
+					Broker: "br",
+				},
+			},
+			brokers: []*eventing.Broker{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "br",
+						Annotations: map[string]string{
+							eventing.BrokerClassAnnotationKey: kafka.BrokerClass,
+						},
+					},
+				},
+			},
+			pass: true,
+		},
+		{
+			name: "exiting non kafka broker",
+			trigger: &eventing.Trigger{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace: "ns",
+					Name:      "tr",
+				},
+				Spec: eventing.TriggerSpec{
+					Broker: "br",
+				},
+			},
+			brokers: []*eventing.Broker{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "br",
+						Annotations: map[string]string{
+							eventing.BrokerClassAnnotationKey: kafka.BrokerClass + "-boh",
+						},
+					},
+				},
+			},
+			pass: false,
+		},
+		{
+			name: "exiting non kafka broker - trigger with kafka broker finalizer",
+			trigger: &eventing.Trigger{
+				ObjectMeta: metav1.ObjectMeta{
+					Namespace:  "ns",
+					Name:       "tr",
+					Finalizers: []string{FinalizerName},
+				},
+				Spec: eventing.TriggerSpec{
+					Broker: "br",
+				},
+			},
+			brokers: []*eventing.Broker{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "ns",
+						Name:      "br",
+						Annotations: map[string]string{
+							eventing.BrokerClassAnnotationKey: kafka.BrokerClass + "-boh",
+						},
+					},
+				},
+			},
+			pass: true,
+		},
+	}
 
-	assert.False(t, pass)
+	for _, tc := range tt {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			brokerInformer := brokerinformer.Get(ctx)
+			for _, obj := range tc.brokers {
+				_ = brokerInformer.Informer().GetStore().Add(obj)
+			}
+			filter := filterTriggers(brokerInformer.Lister())
+			pass := filter(tc.trigger)
+			assert.Equal(t, tc.pass, pass)
+		})
+	}
 }