Repository: knative-sandbox/eventing-kafka-broker
Pull request #2974 (2023-02-21 08:59:22)

[release-1.9] Fix leaking cluster scoped resources in namespaced broker
This is an automated cherry-pick of #2949

```release-note
`KafkaNamespaced` does not leak cluster scoped resources that are in `config-namespaced-broker-resources` anymore
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2974#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign aliok for approval. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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/release-1.9/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":["aliok"]} -->
aliok: /lgtm
/approve
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2974#issuecomment-1438093008" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2974#" title="Author self-approved">knative-prow-robot</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/release-1.9/OWNERS)~~ [aliok]

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

<details>

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

>/cherrypick release-1.8


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>
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2974](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6f8f6a1) into [release-1.9](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/97f9ceeeac4fd59bc93b34ccd5bca475165ce5f1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (97f9cee) will **increase** coverage by `0.26%`.
> The diff coverage is `67.16%`.

```diff
@@               Coverage Diff               @@
##           release-1.9    #2974      +/-   ##
===============================================
+ Coverage        58.57%   58.84%   +0.26%     
===============================================
  Files               78       79       +1     
  Lines             8254     8373     +119     
===============================================
+ Hits              4835     4927      +92     
- Misses            3054     3071      +17     
- Partials           365      375      +10     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ane/pkg/reconciler/broker/namespaced\_controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvbmFtZXNwYWNlZF9jb250cm9sbGVyLmdv) | `0.00% <0.00%> (ø)` | |
| [...l-plane/pkg/reconciler/broker/namespaced\_broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvbmFtZXNwYWNlZF9icm9rZXIuZ28=) | `70.87% <64.70%> (+0.28%)` | :arrow_up: |
| [control-plane/pkg/util/lock\_map.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2974?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvdXRpbC9sb2NrX21hcC5nbw==) | `94.59% <94.59%> (ø)` | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)


Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
index 47324b677..8b6590584 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/200-controller-cluster-role.yaml
@@ -21,6 +21,14 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 rules:
+  - apiGroups:
+      - ""
+    resources:
+      - namespaces
+    verbs:
+      - get
+      - list
+      - watch
   - apiGroups:
       - ""
     resources:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker.go b/control-plane/pkg/reconciler/broker/namespaced_broker.go
index 58ac11d09..befaf78a4 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker.go
@@ -30,6 +30,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	appslisters "k8s.io/client-go/listers/apps/v1"
@@ -42,6 +43,8 @@ import (
 	"knative.dev/pkg/resolver"
 
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
+	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/counter"
@@ -49,6 +52,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/propagator"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 )
 
 // TODO: 5.
@@ -61,11 +65,13 @@ type NamespacedReconciler struct {
 
 	Resolver *resolver.URIResolver
 
+	NamespaceLister          corelisters.NamespaceLister
 	ConfigMapLister          corelisters.ConfigMapLister
 	ServiceAccountLister     corelisters.ServiceAccountLister
 	ServiceLister            corelisters.ServiceLister
 	ClusterRoleBindingLister rbaclisters.ClusterRoleBindingLister
 	DeploymentLister         appslisters.DeploymentLister
+	BrokerLister             eventinglisters.BrokerLister
 
 	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
@@ -78,14 +84,11 @@ type NamespacedReconciler struct {
 
 	IPsLister          prober.IPListerWithMapping
 	ManifestivalClient mf.Client
+
+	DataplaneLifecycleLocksByNamespace util.LockMap[string]
 }
 
 func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
-	r.IPsLister.Register(
-		types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name},
-		prober.GetIPForService(types.NamespacedName{Namespace: broker.Namespace, Name: r.Env.IngressName}),
-	)
-
 	if broker.Spec.Config != nil && broker.Spec.Config.Namespace != "" && broker.Spec.Config.Namespace != broker.Namespace {
 		return propagateErrorCondition(broker,
 			fmt.Errorf("broker with %q class need the config in the same namespace with the broker. broker.spec.config.namespace=%q, broker.namespace=%q",
@@ -94,6 +97,27 @@ func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventi
 
 	br := r.createReconcilerForBrokerInstance(broker)
 
+	event := r.reconcileDataPlane(ctx, broker)
+	if event != nil {
+		return event
+	}
+
+	return br.ReconcileKind(ctx, broker)
+}
+
+// reconcileDataPlane reconciles the data plane in the namespace of the given broker.
+// A lock is acquired for the namespace of the broker to ensure that the creation of the data plane is not done
+// at the same time with the deletion of the data plane.
+func (r *NamespacedReconciler) reconcileDataPlane(ctx context.Context, broker *eventing.Broker) reconciler.Event {
+	namespaceLock := r.DataplaneLifecycleLocksByNamespace.GetLock(broker.Namespace)
+	namespaceLock.Lock()
+	defer namespaceLock.Unlock()
+
+	r.IPsLister.Register(
+		types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name},
+		prober.GetIPForService(types.NamespacedName{Namespace: broker.Namespace, Name: r.Env.IngressName}),
+	)
+
 	manifest, err := r.getManifest(ctx, broker)
 	if err != nil {
 		return propagateErrorCondition(broker, fmt.Errorf("unable to transform dataplane manifest: %w", err))
@@ -103,7 +127,7 @@ func (r *NamespacedReconciler) ReconcileKind(ctx context.Context, broker *eventi
 		return propagateErrorCondition(broker, fmt.Errorf("unable to apply dataplane manifest: %w", err))
 	}
 
-	return br.ReconcileKind(ctx, broker)
+	return nil
 }
 
 func (r *NamespacedReconciler) FinalizeKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -112,9 +136,79 @@ func (r *NamespacedReconciler) FinalizeKind(ctx context.Context, broker *eventin
 
 	r.IPsLister.Unregister(types.NamespacedName{Namespace: broker.Namespace, Name: broker.Name})
 
+	if err := r.finalizeDataPlane(ctx, broker); err != nil {
+		return err
+	}
+
 	return result
 }
 
+// finalizeDataPlane finalizes the data plane in the namespace of the given broker.
+// A lock is acquired for the namespace of the broker to ensure that the deletion of the data plane is not done
+// at the same time with the creation of the data plane.
+func (r *NamespacedReconciler) finalizeDataPlane(ctx context.Context, broker *eventing.Broker) reconciler.Event {
+	namespaceLock := r.DataplaneLifecycleLocksByNamespace.GetLock(broker.Namespace)
+	namespaceLock.Lock()
+	defer namespaceLock.Unlock()
+
+	count, err := r.namespacedBrokerCountInNamespace(broker.Namespace)
+	if err != nil {
+		return fmt.Errorf("failed to count namespaced brokers in the namespace for clean up: %w", err)
+	}
+
+	if count == 0 {
+		// there's no more namespaced broker in the namespace, we can clean up the cluster scoped resources
+		err := r.cleanUpClusterScopedResources(ctx, broker)
+		if err != nil {
+			return fmt.Errorf("failed to clean up cluster scoped resources for KafkaNamespaced brokers in the namespace: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// cleanUpClusterScopedResources deletes the cluster scoped resources created for the data plane in the namespace, if
+// this is the last namespaced broker in the namespace
+func (r *NamespacedReconciler) cleanUpClusterScopedResources(ctx context.Context, broker *eventing.Broker) error {
+	manifest, err := r.getManifest(ctx, broker)
+	if err != nil {
+		return fmt.Errorf("unable to transform dataplane manifest for deletion: %w", err)
+	}
+
+	// Delete the cluster scoped resources created for the data plane in the namespace.
+	manifest = manifest.Filter(FilterClusterScoped)
+
+	// Only delete cluster scoped resources, others are garbage collected by Kubernetes already
+	err = manifest.Delete()
+	if err != nil {
+		return fmt.Errorf("failed to delete additional resources: %w", err)
+	}
+
+	return nil
+}
+
+// namespacedBrokerCountInNamespace returns the number of non-deleted namespaced brokers in the namespace
+func (r *NamespacedReconciler) namespacedBrokerCountInNamespace(namespace string) (int, error) {
+	var list []*eventing.Broker
+	var err error
+
+	if list, err = r.BrokerLister.Brokers(namespace).List(labels.Everything()); err != nil {
+		return 0, fmt.Errorf("failed to list brokers in the namespace: %w", err)
+	}
+
+	namespacedBrokerCount := 0
+	for _, b := range list {
+		if !b.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if b.Annotations[brokerreconciler.ClassAnnotationKey] == kafka.NamespacedBrokerClass {
+			namespacedBrokerCount++
+		}
+	}
+
+	return namespacedBrokerCount, err
+}
+
 func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventing.Broker) *Reconciler {
 
 	return &Reconciler{
@@ -150,6 +244,11 @@ func (r *NamespacedReconciler) createReconcilerForBrokerInstance(broker *eventin
 // getManifest returns the manifest that is transformed from the BaseDataPlaneManifest with the changes
 // for the given broker
 func (r *NamespacedReconciler) getManifest(ctx context.Context, broker *eventing.Broker) (mf.Manifest, error) {
+	namespace, err := r.NamespaceLister.Get(broker.Namespace)
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("failed to get namespace of the broker: %w", err)
+	}
+
 	var resources []unstructured.Unstructured
 
 	additionalConfigMaps, err := r.configMapsFromSystemNamespace(broker)
@@ -203,6 +302,14 @@ func (r *NamespacedReconciler) getManifest(ctx context.Context, broker *eventing
 		return mf.Manifest{}, fmt.Errorf("unable to append owner ref: %w", err)
 	}
 
+	// Set the namespace as owner ref to cluster-scoped resources. We delete them manually, but as a safety net we set
+	// the namespace as owner ref so that even when cluster scoped resources cannot be deleted, they will be garbage
+	// collected when the namespace is deleted.
+	manifest, err = manifest.Transform(appendNamespaceAsOwnerRefsToClusterScoped(namespace))
+	if err != nil {
+		return mf.Manifest{}, fmt.Errorf("unable to set namespace as owner ref to cluster-scoped resources: %w", err)
+	}
+
 	manifest, err = manifest.Transform(filterMetadata)
 	if err != nil {
 		return mf.Manifest{}, fmt.Errorf("unable to filter metadata: %w", err)
@@ -433,6 +540,22 @@ func unstructuredFromObject(obj runtime.Object) (unstructured.Unstructured, erro
 	return unstructured.Unstructured{Object: unstr}, nil
 }
 
+func appendNamespaceAsOwnerRefsToClusterScoped(namespace *corev1.Namespace) mf.Transformer {
+	return func(resource *unstructured.Unstructured) error {
+		if !isClusterScoped(resource.GetKind()) || strings.ToLower(resource.GetKind()) == "namespace" {
+			return nil
+		}
+
+		ref := *metav1.NewControllerRef(namespace, corev1.SchemeGroupVersion.WithKind("Namespace"))
+		ref.Controller = pointer.Bool(true)
+		ref.BlockOwnerDeletion = pointer.Bool(false)
+
+		resource.SetOwnerReferences([]metav1.OwnerReference{ref})
+
+		return nil
+	}
+}
+
 func appendNewOwnerRefsToPersisted(client mf.Client, broker *eventing.Broker) mf.Transformer {
 	return func(resource *unstructured.Unstructured) error {
 		if isClusterScoped(resource.GetKind()) {
@@ -549,3 +672,7 @@ func isClusterScoped(kind string) bool {
 	}
 	return false
 }
+
+func FilterClusterScoped(u *unstructured.Unstructured) bool {
+	return isClusterScoped(u.GetKind())
+}
diff --git a/control-plane/pkg/reconciler/broker/namespaced_controller.go b/control-plane/pkg/reconciler/broker/namespaced_controller.go
index 846aadcac..22b2ec85b 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_controller.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_controller.go
@@ -19,6 +19,9 @@ package broker
 import (
 	"context"
 	"net/http"
+	"time"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 
 	"github.com/Shopify/sarama"
 	mfclient "github.com/manifestival/client-go-client"
@@ -46,6 +49,7 @@ import (
 	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
 	deploymentinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/deployment"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
+	namespaceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/namespace"
 	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
 	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
 	serviceinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/service"
@@ -89,15 +93,18 @@ func NewNamespacedController(ctx context.Context, watcher configmap.Watcher, env
 			DispatcherLabel:              base.BrokerDispatcherLabel,
 			ReceiverLabel:                base.BrokerReceiverLabel,
 		},
-		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
-		ConfigMapLister:            configmapInformer.Lister(),
-		ServiceAccountLister:       serviceaccountinformer.Get(ctx).Lister(),
-		ServiceLister:              serviceinformer.Get(ctx).Lister(),
-		ClusterRoleBindingLister:   clusterrolebindinginformer.Get(ctx).Lister(),
-		DeploymentLister:           deploymentinformer.Get(ctx).Lister(),
-		Env:                        env,
-		Counter:                    counter.NewExpiringCounter(ctx),
-		ManifestivalClient:         mfc,
+		NewKafkaClusterAdminClient:         sarama.NewClusterAdmin,
+		NamespaceLister:                    namespaceinformer.Get(ctx).Lister(),
+		ConfigMapLister:                    configmapInformer.Lister(),
+		ServiceAccountLister:               serviceaccountinformer.Get(ctx).Lister(),
+		ServiceLister:                      serviceinformer.Get(ctx).Lister(),
+		ClusterRoleBindingLister:           clusterrolebindinginformer.Get(ctx).Lister(),
+		DeploymentLister:                   deploymentinformer.Get(ctx).Lister(),
+		BrokerLister:                       brokerinformer.Get(ctx).Lister(),
+		Env:                                env,
+		Counter:                            counter.NewExpiringCounter(ctx),
+		ManifestivalClient:                 mfc,
+		DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.NamespacedBrokerClass, func(impl *controller.Impl) controller.Options {
diff --git a/control-plane/pkg/util/lock_map.go b/control-plane/pkg/util/lock_map.go
new file mode 100644
index 000000000..a1c9b6e8d
--- /dev/null
+++ b/control-plane/pkg/util/lock_map.go
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2023 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 util
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// LockMap provides a mechanism to get a lock for a given key.
+type LockMap[K comparable] interface {
+	GetLock(K) *sync.Mutex
+}
+
+type expiringLockMap[K comparable] struct {
+	mu     sync.Mutex
+	locks  map[K]*sync.Mutex
+	access map[K]time.Time
+
+	ttl time.Duration
+}
+
+// NewExpiringLockMap returns a new LockMap that removes entries after the given TTL. Timing of the removal is done
+// in a best-effort fashion, and is not guaranteed to be exact.
+func NewExpiringLockMap[K comparable](ctx context.Context, ttl time.Duration) LockMap[K] {
+	lm := &expiringLockMap[K]{
+		locks:  make(map[K]*sync.Mutex),
+		access: make(map[K]time.Time),
+		ttl:    ttl,
+	}
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(ttl):
+				lm.removeExpiredEntries(time.Now())
+			}
+		}
+	}()
+
+	return lm
+}
+
+func (lm *expiringLockMap[K]) GetLock(key K) *sync.Mutex {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	var lock *sync.Mutex
+	var ok bool
+	if lock, ok = lm.locks[key]; !ok {
+		lock = &sync.Mutex{}
+		lm.locks[key] = lock
+	}
+
+	lm.access[key] = time.Now()
+
+	return lock
+}
+
+func (lm *expiringLockMap[K]) removeExpiredEntries(now time.Time) {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	for key, lastAccess := range lm.access {
+		if now.Sub(lastAccess) > lm.ttl {
+			delete(lm.locks, key)
+			delete(lm.access, key)
+		}
+	}
+}
diff --git a/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go
new file mode 100644
index 000000000..aa57e789f
--- /dev/null
+++ b/vendor/knative.dev/pkg/client/injection/kube/informers/core/v1/namespace/namespace.go
@@ -0,0 +1,110 @@
+/*
+Copyright 2022 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.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package namespace
+
+import (
+	context "context"
+
+	apicorev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	v1 "k8s.io/client-go/informers/core/v1"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1 "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+	client "knative.dev/pkg/client/injection/kube/client"
+	factory "knative.dev/pkg/client/injection/kube/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Core().V1().Namespaces()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx), resourceVersion: injection.GetResourceVersion(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1.NamespaceInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch k8s.io/client-go/informers/core/v1.NamespaceInformer from context.")
+	}
+	return untyped.(v1.NamespaceInformer)
+}
+
+type wrapper struct {
+	client kubernetes.Interface
+
+	resourceVersion string
+}
+
+var _ v1.NamespaceInformer = (*wrapper)(nil)
+var _ corev1.NamespaceLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apicorev1.Namespace{}, 0, nil)
+}
+
+func (w *wrapper) Lister() corev1.NamespaceLister {
+	return w
+}
+
+// SetResourceVersion allows consumers to adjust the minimum resourceVersion
+// used by the underlying client.  It is not accessible via the standard
+// lister interface, but can be accessed through a user-defined interface and
+// an implementation check e.g. rvs, ok := foo.(ResourceVersionSetter)
+func (w *wrapper) SetResourceVersion(resourceVersion string) {
+	w.resourceVersion = resourceVersion
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apicorev1.Namespace, err error) {
+	lo, err := w.client.CoreV1().Namespaces().List(context.TODO(), metav1.ListOptions{
+		LabelSelector:   selector.String(),
+		ResourceVersion: w.resourceVersion,
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apicorev1.Namespace, error) {
+	return w.client.CoreV1().Namespaces().Get(context.TODO(), name, metav1.GetOptions{
+		ResourceVersion: w.resourceVersion,
+	})
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 178cf5a95..84d32b553 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1401,6 +1401,7 @@ knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset
 knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/configmap
 knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake
+knative.dev/pkg/client/injection/kube/informers/core/v1/namespace
 knative.dev/pkg/client/injection/kube/informers/core/v1/node
 knative.dev/pkg/client/injection/kube/informers/core/v1/node/fake
 knative.dev/pkg/client/injection/kube/informers/core/v1/pod


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
index 848f90259..7da4b1e5d 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -20,8 +20,10 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/utils/pointer"
@@ -34,10 +36,12 @@ import (
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/util"
 
 	"github.com/Shopify/sarama"
 	"github.com/manifestival/client-go-client"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
@@ -79,6 +83,9 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 		{
 			Name: "Reconciled normal",
 			Objects: []runtime.Object{
+				reconcilertesting.NewNamespace(BrokerNamespace, func(ns *corev1.Namespace) {
+					ns.UID = BrokerNamespaceUUID
+				}),
 				NewNamespacedBroker(
 					WithBrokerConfig(
 						KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
@@ -108,6 +115,24 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
 					WithClusterRoleBindingRoleRef("knative-kafka-broker-data-plane"),
 				),
+				NewConfigMapWithTextData(SystemNamespace, NamespacedBrokerAdditionalResourcesConfigMapName, map[string]string{
+					"resources": `
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: ClusterRole
+  metadata:
+    name: test-role
+    labels:
+      "knative.foo": "foo-{{.Namespace}}"
+    creationTimestamp: null
+  rules:
+    - apiGroups:
+        - "v1"
+      resources:
+        - pods
+      verbs:
+        - get
+`,
+				}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -164,6 +189,16 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					WithNamespacedBrokerOwnerRef,
 					WithNamespacedLabel,
 				),
+				ToManifestivalResource(t, NewClusterRole("test-role",
+					WithClusterRoleLabel("knative.foo", "foo-"+BrokerNamespace),
+					WithClusterRoleRules(rbacv1.PolicyRule{
+						APIGroups: []string{"v1"},
+						Resources: []string{"pods"},
+						Verbs:     []string{"get"},
+					})),
+					WithNamespacedBrokerNamespaceAsOwnerRef,
+					WithNamespacedLabel,
+				),
 				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil,
 					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
 					WithConfigmapOwnerRef(&metav1.OwnerReference{
@@ -230,6 +265,9 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 					),
 				},
 			},
+			// true since we're creating cluster scoped resources and namespace validation is checking if the namespace
+			// of the broker is the same with the namespace of the cluster scoped resource (which is nil)
+			SkipNamespaceValidation: true,
 		},
 	}
 
@@ -258,6 +296,9 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 		{
 			Name: "Reconciled normal",
 			Objects: []runtime.Object{
+				reconcilertesting.NewNamespace(BrokerNamespace, func(ns *corev1.Namespace) {
+					ns.UID = BrokerNamespaceUUID
+				}),
 				NewDeletedBroker(reconcilertesting.WithBrokerClass(kafka.NamespacedBrokerClass)),
 				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
@@ -270,6 +311,41 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 					},
 					Generation: 1,
 				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+				reconcilertesting.NewConfigMap(env.DataPlaneConfigConfigMapName, SystemNamespace),
+				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
+				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-receiver", SystemNamespace),
+				reconcilertesting.NewDeployment("kafka-broker-dispatcher", SystemNamespace),
+				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
+				NewClusterRoleBinding("knative-kafka-broker-data-plane",
+					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+				),
+				NewConfigMapWithTextData(SystemNamespace, NamespacedBrokerAdditionalResourcesConfigMapName, map[string]string{
+					"resources": `
+- apiVersion: rbac.authorization.k8s.io/v1
+  kind: ClusterRole
+  metadata:
+    name: test-role
+    labels:
+      "knative.foo": "foo-{{.Namespace}}"
+    creationTimestamp: null
+  rules:
+    - apiGroups:
+        - "v1"
+      resources:
+        - pods
+      verbs:
+        - get
+`,
+				}),
+				NewClusterRole("test-role",
+					WithClusterRoleLabel("knative.foo", "foo-"+BrokerNamespace),
+					WithClusterRoleRules(rbacv1.PolicyRule{
+						APIGroups: []string{"v1"},
+						Resources: []string{"pods"},
+						Verbs:     []string{"get"},
+					})),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
@@ -285,6 +361,18 @@ func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
 					}),
 				),
 			},
+			WantDeletes: []clientgotesting.DeleteActionImpl{
+				{
+					ActionImpl: clientgotesting.ActionImpl{
+						Resource: schema.GroupVersionResource{
+							Group:    rbacv1.SchemeGroupVersion.Group,
+							Version:  rbacv1.SchemeGroupVersion.Version,
+							Resource: "clusterroles",
+						},
+					},
+					Name: "test-role",
+				},
+			},
 			OtherTestData: map[string]interface{}{
 				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
@@ -354,8 +442,10 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 				ReceiverLabel:               base.BrokerReceiverLabel,
 				Tracker:                     &FakeTracker{},
 			},
+			NamespaceLister:          listers.GetNamespaceLister(),
 			ConfigMapLister:          listers.GetConfigMapLister(),
 			DeploymentLister:         listers.GetDeploymentLister(),
+			BrokerLister:             listers.GetBrokerLister(),
 			ServiceAccountLister:     listers.GetServiceAccountLister(),
 			ServiceLister:            listers.GetServiceLister(),
 			ClusterRoleBindingLister: listers.GetClusterRoleBindingLister(),
@@ -370,9 +460,10 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 					T:                                      t,
 				}, nil
 			},
-			Env:                env,
-			Prober:             proberMock,
-			ManifestivalClient: mfcMockClient,
+			Env:                                env,
+			Prober:                             proberMock,
+			ManifestivalClient:                 mfcMockClient,
+			DataplaneLifecycleLocksByNamespace: util.NewExpiringLockMap[string](ctx, time.Minute*30),
 		}
 
 		r := brokerreconciler.NewReconciler(
@@ -408,6 +499,22 @@ func WithNamespacedBrokerOwnerRef(u *unstructured.Unstructured) {
 	u.SetOwnerReferences(refs)
 }
 
+func WithNamespacedBrokerNamespaceAsOwnerRef(u *unstructured.Unstructured) {
+	refs := u.GetOwnerReferences()
+	if refs == nil {
+		refs = []metav1.OwnerReference{}
+	}
+	refs = append(refs, metav1.OwnerReference{
+		APIVersion:         corev1.SchemeGroupVersion.String(),
+		Kind:               "Namespace",
+		Name:               BrokerNamespace,
+		UID:                BrokerNamespaceUUID,
+		Controller:         pointer.Bool(true),
+		BlockOwnerDeletion: pointer.Bool(false),
+	})
+	u.SetOwnerReferences(refs)
+}
+
 func WithNamespacedLabel(u *unstructured.Unstructured) {
 	labels := u.GetLabels()
 	if labels == nil {
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 52ce9cb08..b6d57ea04 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -164,6 +164,10 @@ func (l *Listers) GetClusterRoleBindingLister() rbaclisters.ClusterRoleBindingLi
 	return rbaclisters.NewClusterRoleBindingLister(l.indexerFor(&rbacv1.ClusterRoleBinding{}))
 }
 
+func (l *Listers) GetNamespaceLister() corelisters.NamespaceLister {
+	return corelisters.NewNamespaceLister(l.indexerFor(&corev1.Namespace{}))
+}
+
 func (l *Listers) indexerFor(obj runtime.Object) cache.Indexer {
 	return l.sorter.IndexerForObjectType(obj)
 }
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index c94d3ea45..8b1896289 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -43,10 +43,11 @@ import (
 )
 
 const (
-	BrokerUUID        = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
-	BrokerNamespace   = "test-namespace"
-	BrokerName        = "test-broker"
-	ExternalTopicName = "test-topic"
+	BrokerUUID          = "e7185016-5d98-4b54-84e8-3b1cd4acc6b4"
+	BrokerNamespaceUUID = "d1234567-8910-1234-5678-901234567890"
+	BrokerNamespace     = "test-namespace"
+	BrokerName          = "test-broker"
+	ExternalTopicName   = "test-topic"
 
 	SecretFinalizerName = "kafka.eventing/" + BrokerUUID
 
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 88605d227..30bcf8633 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -393,6 +393,40 @@ func NewKedaSecret(ns, name string) *corev1.Secret {
 	}
 }
 
+// ClusterRoleOption enables further configuration of a ClusterRole.
+type ClusterRoleOption func(role *rbacv1.ClusterRole)
+
+func NewClusterRole(name string, cro ...ClusterRoleOption) *rbacv1.ClusterRole {
+	cr := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+
+	for _, opt := range cro {
+		opt(cr)
+	}
+
+	return cr
+}
+
+// WithClusterRoleRules is a ClusterRoleOption that adds the given rules to the ClusterRole.
+func WithClusterRoleRules(rules ...rbacv1.PolicyRule) ClusterRoleOption {
+	return func(role *rbacv1.ClusterRole) {
+		role.Rules = append(role.Rules, rules...)
+	}
+}
+
+// WithClusterRoleLabel is a ClusterRoleOption that adds the given labels to the ClusterRole.
+func WithClusterRoleLabel(key, value string) ClusterRoleOption {
+	return func(role *rbacv1.ClusterRole) {
+		if role.Labels == nil {
+			role.Labels = make(map[string]string)
+		}
+		role.Labels[key] = value
+	}
+}
+
 func loadCerts() (ca, userKey, userCert []byte) {
 	ca, err := os.ReadFile("testdata/ca.crt")
 	if err != nil {
diff --git a/control-plane/pkg/util/lock_map_test.go b/control-plane/pkg/util/lock_map_test.go
new file mode 100644
index 000000000..6675279f1
--- /dev/null
+++ b/control-plane/pkg/util/lock_map_test.go
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2023 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 util
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestExpiringLockMap_GetLock(t *testing.T) {
+	chanCount := 1000 // controls concurrency
+	size := 50        // controls how many keys are used and thus the collision rate
+
+	r := rand.New(rand.NewSource(42))
+	randomKey := func() string {
+		return fmt.Sprintf("key-%d", r.Intn(size))
+	}
+
+	wg := sync.WaitGroup{}
+	chans := make([]chan string, chanCount)
+
+	// fill channels with data
+	for i := 0; i < chanCount; i++ {
+		chans[i] = make(chan string, size)
+		for j := 0; j < size; j++ {
+			chans[i] <- randomKey()
+		}
+		close(chans[i])
+	}
+	wg.Add(chanCount)
+
+	// consume data from channels concurrently. Lock and unlock the lock for each key.
+	// TTL is 10 hours, so the locks should never expire.
+	lm := NewExpiringLockMap[string](context.Background(), 10*time.Hour)
+
+	for i := 0; i < chanCount; i++ {
+		go func(i int) {
+			defer wg.Done()
+			for key := range chans[i] {
+				func() {
+					lock := lm.GetLock(key)
+					lock.Lock()
+					defer lock.Unlock()
+				}()
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
+func TestExpiringLockMapExpiration(t *testing.T) {
+	lm := NewExpiringLockMap[string](context.Background(), 100*time.Millisecond)
+
+	l1_a := lm.GetLock("l1")
+
+	time.Sleep(50 * time.Millisecond)
+	l1_b := lm.GetLock("l1") // should be the same lock as l1_a, as TTL is 100ms and 50 < 100
+
+	time.Sleep(300 * time.Millisecond)
+	l1_c := lm.GetLock("l1") // should be a different lock as TTL is 100ms and 300 > 100
+
+	if l1_a != l1_b {
+		t.Errorf("l1_a != l1_b")
+	}
+	if l1_a == l1_c {
+		t.Errorf("l1_a == l1_c")
+	}
+}