Repository: knative-sandbox/eventing-kafka-broker
Pull request #2819 (2022-12-02 10:07:06)

[release-1.8] Missing unit tests for namespaced broker
This is an automated cherry-pick of #2817

```release-note

```

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/2819#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign psschwei for approval by writing `/assign @psschwei` in a comment. 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.8/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":["psschwei"]} -->
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/2819#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2819#pullrequestreview-1202262861" title="Approved">matzew</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.8/OWNERS)~~ [matzew]

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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2819](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f33921c) into [release-1.8](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/cd3751eb04a3e60a9c1f49f4e7dc693407c9bbf2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (cd3751e) will **increase** coverage by `2.84%`.
> The diff coverage is `51.80%`.

```diff
@@               Coverage Diff               @@
##           release-1.8    #2819      +/-   ##
===============================================
+ Coverage        55.19%   58.04%   +2.84%     
===============================================
  Files               75       75              
  Lines             8035     8075      +40     
===============================================
+ Hits              4435     4687     +252     
+ Misses            3276     3031     -245     
- Partials           324      357      +33     
```


| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819?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/reconciler/base/reconciler.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY29uY2lsZXIuZ28=) | `61.77% <0.00%> (-4.42%)` | :arrow_down: |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `53.04% <56.36%> (+0.15%)` | :arrow_up: |
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | `61.49% <90.00%> (+1.23%)` | :arrow_up: |
| [control-plane/pkg/apis/core/pod\_defaulting.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9jb3JlL3BvZF9kZWZhdWx0aW5nLmdv) | `87.75% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `47.98% <100.00%> (-2.19%)` | :arrow_down: |
| [...l-plane/pkg/reconciler/broker/namespaced\_broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2819/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvbmFtZXNwYWNlZF9icm9rZXIuZ28=) | `72.15% <0.00%> (+72.15%)` | :arrow_up: |

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)

knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/reconciler-tests_eventing-kafka-broker_release-1.8/1598619768384917504) | true | `/test reconciler-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
knative-prow[bot]: @knative-prow-robot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/reconciler-tests_eventing-kafka-broker_release-1.8/1598619768384917504) | true | `/test reconciler-tests`
channel-integration-tests-ssl_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/channel-integration-tests-ssl_eventing-kafka-broker_release-1.8/1598619768514940928) | true | `/test channel-integration-tests-ssl`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
knative-prow[bot]: @knative-prow-robot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
reconciler-tests_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/reconciler-tests_eventing-kafka-broker_release-1.8/1598619768384917504) | true | `/test reconciler-tests`
channel-integration-tests-ssl_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/channel-integration-tests-ssl_eventing-kafka-broker_release-1.8/1598619768514940928) | true | `/test channel-integration-tests-ssl`
channel-reconciler-tests-sasl-plain_eventing-kafka-broker_release-1.8 | f33921c34f47ce491afc0b5aa65931a5c991d5fc | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2819/channel-reconciler-tests-sasl-plain_eventing-kafka-broker_release-1.8/1598619770528206848) | true | `/test channel-reconciler-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
pierDipi: /retest-required

Patch:
diff --git a/vendor/github.com/manifestival/manifestival/fake/client.go b/vendor/github.com/manifestival/manifestival/fake/client.go
new file mode 100644
index 000000000..60e9c31be
--- /dev/null
+++ b/vendor/github.com/manifestival/manifestival/fake/client.go
@@ -0,0 +1,102 @@
+package fake
+
+import (
+	"fmt"
+
+	mf "github.com/manifestival/manifestival"
+	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+var _ mf.Client = &Client{}
+
+// A convenient way to stub out a Client for test fixtures. Default
+// behavior does nothing and returns a nil error.
+type Client struct {
+	Stubs
+}
+
+// Override any of the Client functions
+type Stubs struct {
+	Create mutator
+	Update mutator
+	Delete mutator
+	Get    accessor
+}
+
+type mutator func(obj *unstructured.Unstructured) error
+type accessor func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error)
+
+// New returns a fully-functioning Client, "persisting" resources in a
+// map, optionally initialized with some API objects
+func New(objs ...runtime.Object) Client {
+	store := map[string]*unstructured.Unstructured{}
+	key := func(u *unstructured.Unstructured) string {
+		return fmt.Sprintf("%s, %s/%s", u.GroupVersionKind(), u.GetNamespace(), u.GetName())
+	}
+	for _, obj := range objs {
+		u := &unstructured.Unstructured{}
+		if err := scheme.Scheme.Convert(obj, u, nil); err != nil {
+			panic(err)
+		}
+		store[key(u)] = u
+	}
+	apply := func(u *unstructured.Unstructured) error {
+		store[key(u)] = u
+		return nil
+	}
+	return Client{
+		Stubs{
+			Create: apply,
+			Update: apply,
+			Delete: func(u *unstructured.Unstructured) error {
+				delete(store, key(u))
+				return nil
+			},
+			Get: func(u *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+				v, found := store[key(u)]
+				if !found {
+					gvk := u.GroupVersionKind()
+					gr := schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}
+					return nil, errors.NewNotFound(gr, u.GetName())
+				}
+				return v, nil
+			},
+		},
+	}
+}
+
+// Manifestival.Client.Create
+func (c Client) Create(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
+	if c.Stubs.Create != nil {
+		return c.Stubs.Create(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Update
+func (c Client) Update(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
+	if c.Stubs.Update != nil {
+		return c.Stubs.Update(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Delete
+func (c Client) Delete(obj *unstructured.Unstructured, options ...mf.DeleteOption) error {
+	if c.Stubs.Delete != nil {
+		return c.Stubs.Delete(obj)
+	}
+	return nil
+}
+
+// Manifestival.Client.Get
+func (c Client) Get(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	if c.Stubs.Get != nil {
+		return c.Stubs.Get(obj)
+	}
+	return nil, nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index d7b1dcf30..97de57bb8 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -329,6 +329,7 @@ github.com/manifestival/client-go-client/pkg/dynamic
 # github.com/manifestival/manifestival v0.7.1
 ## explicit; go 1.15
 github.com/manifestival/manifestival
+github.com/manifestival/manifestival/fake
 github.com/manifestival/manifestival/internal/overlay
 github.com/manifestival/manifestival/internal/patch
 github.com/manifestival/manifestival/internal/sources


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 19a161ecb..525c6a40f 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -90,13 +90,14 @@ var (
 )
 
 var DefaultEnv = &config.Env{
-	DataPlaneConfigMapNamespace: "knative-eventing",
-	ContractConfigMapName:       "kafka-broker-brokers-triggers",
-	GeneralConfigMapName:        "kafka-broker-config",
-	IngressName:                 "kafka-broker-ingress",
-	SystemNamespace:             "knative-eventing",
-	ContractConfigMapFormat:     base.Json,
-	DefaultBackoffDelayMs:       1000,
+	DataPlaneConfigMapNamespace:  "knative-eventing",
+	DataPlaneConfigConfigMapName: "config-kafka-broker-data-plane",
+	ContractConfigMapName:        "kafka-broker-brokers-triggers",
+	GeneralConfigMapName:         "kafka-broker-config",
+	IngressName:                  "kafka-broker-ingress",
+	SystemNamespace:              "knative-eventing",
+	ContractConfigMapFormat:      base.Json,
+	DefaultBackoffDelayMs:        1000,
 }
 
 func TestBrokerReconciler(t *testing.T) {
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
new file mode 100644
index 000000000..7d8695076
--- /dev/null
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -0,0 +1,320 @@
+/*
+ * Copyright 2020 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 broker_test // different package name due to import cycles. (broker -> testing -> broker)
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	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"
+
+	mffake "github.com/manifestival/manifestival/fake"
+
+	"github.com/Shopify/sarama"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	clientgotesting "k8s.io/client-go/testing"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/logging"
+	. "knative.dev/pkg/reconciler/testing"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
+	fakeeventingclient "knative.dev/eventing/pkg/client/injection/client/fake"
+	brokerreconciler "knative.dev/eventing/pkg/client/injection/reconciler/eventing/v1/broker"
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+)
+
+func TestNamespacedBrokerReconciler(t *testing.T) {
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
+
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerReconciliation(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewNamespacedBroker(
+					WithBrokerConfig(
+						KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+					),
+				),
+				BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace)),
+				DataPlaneConfigMap(SystemNamespace, env.DataPlaneConfigConfigMapName, ConsumerConfigKey,
+					DataPlaneConfigInitialOffset(ConsumerConfigKey, sources.OffsetLatest),
+				),
+				reconcilertesting.NewConfigMap("config-tracing", SystemNamespace),
+				reconcilertesting.NewConfigMap("kafka-config-logging", SystemNamespace),
+				NewConfigMapWithBinaryData(env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, nil),
+				NewService(),
+				BrokerReceiverPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPod(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				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",
+					WithSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane", "knative-kafka-broker-data-plane"),
+				),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(BrokerNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat,
+					&contract.Contract{
+						Resources: []*contract.Resource{
+							{
+								Uid:              BrokerUUID,
+								Topics:           []string{BrokerTopic()},
+								Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+								BootstrapServers: bootstrapServers,
+								Reference:        BrokerReference(),
+							},
+						},
+						Generation: 1,
+					},
+					reconcilertesting.WithConfigMapLabels(metav1.LabelSelector{MatchLabels: map[string]string{"eventing.knative.dev/namespaced": "true"}}),
+					WithConfigmapOwnerRef(&metav1.OwnerReference{
+						APIVersion:         eventing.SchemeGroupVersion.String(),
+						Kind:               "Broker",
+						Name:               BrokerName,
+						UID:                BrokerUUID,
+						Controller:         pointer.Bool(false),
+						BlockOwnerDeletion: pointer.Bool(true),
+					}),
+				),
+				BrokerReceiverPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				BrokerDispatcherPodUpdate(BrokerNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewNamespacedBroker(
+						reconcilertesting.WithInitBrokerConditions,
+						WithBrokerConfig(
+							KReference(BrokerConfig(bootstrapServers, 20, 5, WithConfigMapNamespace(BrokerNamespace))),
+						),
+						StatusBrokerConfigMapUpdatedReady(&env),
+						StatusBrokerDataPlaneAvailable,
+						StatusBrokerConfigParsed,
+						StatusBrokerTopicReady,
+						NamespacedBrokerAddressable(&env),
+						StatusBrokerProbeSucceeded,
+						BrokerConfigMapAnnotations(),
+						WithTopicStatusAnnotation(BrokerTopic()),
+					),
+				},
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func TestNamespacedBrokerFinalizer(t *testing.T) {
+	t.Parallel()
+
+	for _, f := range Formats {
+		namespacedBrokerFinalization(t, f, *DefaultEnv)
+	}
+}
+
+func namespacedBrokerFinalization(t *testing.T, format string, env config.Env) {
+
+	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
+
+	env.ContractConfigMapFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal",
+			Objects: []runtime.Object{
+				NewDeletedBroker(reconcilertesting.WithBrokerClass(kafka.NamespacedBrokerClass)),
+				BrokerConfig(bootstrapServers, 20, 5),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     BrokerUUID,
+							Topics:  []string{BrokerTopic()},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
+						},
+					},
+					Generation: 1,
+				}, env.DataPlaneConfigMapNamespace, env.ContractConfigMapName, env.ContractConfigMapFormat),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
+			},
+			OtherTestData: map[string]interface{}{
+				testProber: probertesting.MockProber(prober.StatusNotReady),
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTableNamespaced(t, table, &env)
+}
+
+func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
+
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
+
+		defaultTopicDetail := sarama.TopicDetail{
+			NumPartitions:     DefaultNumPartitions,
+			ReplicationFactor: DefaultReplicationFactor,
+		}
+
+		var onCreateTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnCreateTopic]; ok {
+			onCreateTopicError = want.(error)
+		}
+
+		var onDeleteTopicError error
+		if want, ok := row.OtherTestData[wantErrorOnDeleteTopic]; ok {
+			onDeleteTopicError = want.(error)
+		}
+
+		expectedTopicDetail := defaultTopicDetail
+		if td, ok := row.OtherTestData[ExpectedTopicDetail]; ok {
+			expectedTopicDetail = td.(sarama.TopicDetail)
+		}
+
+		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName)
+		if t, ok := row.OtherTestData[externalTopic]; ok {
+			expectedTopicName = t.(string)
+		}
+
+		var metadata []*sarama.TopicMetadata
+		metadata = append(metadata, &sarama.TopicMetadata{
+			Name:       ExternalTopicName,
+			IsInternal: false,
+			Partitions: []*sarama.PartitionMetadata{{}},
+		})
+
+		proberMock := probertesting.MockProber(prober.StatusReady)
+		if p, ok := row.OtherTestData[testProber]; ok {
+			proberMock = p.(prober.Prober)
+		}
+
+		mfcMockClient := mffake.New()
+
+		reconciler := &NamespacedReconciler{
+			Reconciler: &base.Reconciler{
+				KubeClient:                  kubeclient.Get(ctx),
+				PodLister:                   listers.GetPodLister(),
+				SecretLister:                listers.GetSecretLister(),
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				ContractConfigMapName:       env.ContractConfigMapName,
+				ContractConfigMapFormat:     env.ContractConfigMapFormat,
+				DataPlaneNamespace:          env.SystemNamespace,
+				DispatcherLabel:             base.BrokerDispatcherLabel,
+				ReceiverLabel:               base.BrokerReceiverLabel,
+			},
+			ConfigMapLister:          listers.GetConfigMapLister(),
+			DeploymentLister:         listers.GetDeploymentLister(),
+			ServiceAccountLister:     listers.GetServiceAccountLister(),
+			ServiceLister:            listers.GetServiceLister(),
+			ClusterRoleBindingLister: listers.GetClusterRoleBindingLister(),
+			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopicName:                      expectedTopicName,
+					ExpectedTopicDetail:                    expectedTopicDetail,
+					ErrorOnCreateTopic:                     onCreateTopicError,
+					ErrorOnDeleteTopic:                     onDeleteTopicError,
+					ExpectedTopics:                         []string{expectedTopicName},
+					ExpectedTopicsMetadataOnDescribeTopics: metadata,
+					T:                                      t,
+				}, nil
+			},
+			Env:                env,
+			Prober:             proberMock,
+			ManifestivalClient: mfcMockClient,
+		}
+
+		reconciler.ConfigMapTracker = &FakeTracker{}
+		reconciler.SecretTracker = &FakeTracker{}
+
+		r := brokerreconciler.NewReconciler(
+			ctx,
+			logging.FromContext(ctx),
+			fakeeventingclient.Get(ctx),
+			listers.GetBrokerLister(),
+			controller.GetEventRecorder(ctx),
+			reconciler,
+			kafka.NamespacedBrokerClass,
+		)
+
+		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
+		reconciler.IPsLister = prober.NewIPListerWithMapping()
+
+		return r
+	}))
+}
diff --git a/control-plane/pkg/reconciler/testing/listers.go b/control-plane/pkg/reconciler/testing/listers.go
index 31a261c78..7169afed7 100644
--- a/control-plane/pkg/reconciler/testing/listers.go
+++ b/control-plane/pkg/reconciler/testing/listers.go
@@ -17,11 +17,15 @@
 package testing
 
 import (
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
 	fakeapiextensionsclientset "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
 	"k8s.io/apimachinery/pkg/runtime"
 	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	appslisters "k8s.io/client-go/listers/apps/v1"
 	corelisters "k8s.io/client-go/listers/core/v1"
+	rbaclisters "k8s.io/client-go/listers/rbac/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	messaging "knative.dev/eventing/pkg/apis/messaging/v1"
@@ -114,6 +118,10 @@ func (l *Listers) GetPodLister() corelisters.PodLister {
 	return corelisters.NewPodLister(l.indexerFor(&corev1.Pod{}))
 }
 
+func (l *Listers) GetDeploymentLister() appslisters.DeploymentLister {
+	return appslisters.NewDeploymentLister(l.indexerFor(&appsv1.Deployment{}))
+}
+
 func (l *Listers) GetSecretLister() corelisters.SecretLister {
 	return corelisters.NewSecretLister(l.indexerFor(&corev1.Secret{}))
 }
@@ -134,6 +142,10 @@ func (l *Listers) GetServiceLister() corelisters.ServiceLister {
 	return corelisters.NewServiceLister(l.indexerFor(&corev1.Service{}))
 }
 
+func (l *Listers) GetServiceAccountLister() corelisters.ServiceAccountLister {
+	return corelisters.NewServiceAccountLister(l.indexerFor(&corev1.ServiceAccount{}))
+}
+
 func (l *Listers) GetKafkaSinkLister() eventingkafkabrokerlisters.KafkaSinkLister {
 	return eventingkafkabrokerlisters.NewKafkaSinkLister(l.indexerFor(&eventingkafkabroker.KafkaSink{}))
 }
@@ -154,6 +166,10 @@ func (l *Listers) GetConsumerLister() consumerlisters.ConsumerLister {
 	return consumerlisters.NewConsumerLister(l.indexerFor(&eventingkafkabrokerconsumer.Consumer{}))
 }
 
+func (l *Listers) GetClusterRoleBindingLister() rbaclisters.ClusterRoleBindingLister {
+	return rbaclisters.NewClusterRoleBindingLister(l.indexerFor(&rbacv1.ClusterRoleBinding{}))
+}
+
 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 51c821832..ad618f94a 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -192,6 +192,12 @@ func WithBootstrapServerStatusAnnotation(servers string) reconcilertesting.Broke
 
 type CMOption func(cm *corev1.ConfigMap)
 
+func WithConfigMapNamespace(namespace string) CMOption {
+	return func(cm *corev1.ConfigMap) {
+		cm.ObjectMeta.Namespace = namespace
+	}
+}
+
 func BrokerConfig(bootstrapServers string, numPartitions, replicationFactor int, options ...CMOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
@@ -272,17 +278,25 @@ func StatusBrokerConfigNotParsed(reason string) func(broker *eventing.Broker) {
 }
 
 func BrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
-
 	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, env.SystemNamespace)
+	}
+}
 
-		broker.Status.Address.URL = &apis.URL{
-			Scheme: "http",
-			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
-			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
-		}
+func NamespacedBrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
+	return func(broker *eventing.Broker) {
+		brokerAddressable(broker, env.IngressName, broker.Namespace)
+	}
+}
 
-		broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
+func brokerAddressable(broker *eventing.Broker, serviceName, serviceNamespace string) {
+	broker.Status.Address.URL = &apis.URL{
+		Scheme: "http",
+		Host:   network.GetServiceHostname(serviceName, serviceNamespace),
+		Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
 	}
+
+	broker.GetConditionSet().Manage(&broker.Status).MarkTrue(base.ConditionAddressable)
 }
 
 func BrokerReference() *contract.Reference {
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index b279d4f23..5667432a1 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -158,6 +158,39 @@ func NewRoleBinding(namespace string, name string, mutations ...func(account *rb
 	return s
 }
 
+func WithSubjectServiceAccount(saNamespace, saName, roleName string) func(account *rbacv1.ClusterRoleBinding) {
+	return func(cb *rbacv1.ClusterRoleBinding) {
+		cb.Subjects = []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: saNamespace,
+				Name:      saName,
+			},
+		}
+		cb.RoleRef = rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     roleName,
+			APIGroup: "rbac.authorization.k8s.io",
+		}
+	}
+}
+
+func NewClusterRoleBinding(name string, mutations ...func(account *rbacv1.ClusterRoleBinding)) *rbacv1.ClusterRoleBinding {
+	s := &rbacv1.ClusterRoleBinding{
+		TypeMeta: metav1.TypeMeta{
+			Kind:       "ClusterRoleBinding",
+			APIVersion: rbacv1.SchemeGroupVersion.String(),
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	for _, mut := range mutations {
+		mut(s)
+	}
+	return s
+}
+
 func NewConfigMapWithBinaryData(namespace string, name string, data []byte, options ...reconcilertesting.ConfigMapOption) runtime.Object {
 	return reconcilertesting.NewConfigMap(
 		name,
diff --git a/control-plane/pkg/reconciler/testing/objects_consumergroup.go b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
index 4f776743a..da91ead8c 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumergroup.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumergroup.go
@@ -19,6 +19,8 @@ package testing
 import (
 	"time"
 
+	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -227,3 +229,9 @@ func NewDeletedConsumeGroup(opts ...ConsumerGroupOption) runtime.Object {
 func WithDeletedTimeStampConsumeGroup(cg *kafkainternals.ConsumerGroup) {
 	cg.GetObjectMeta().SetDeletionTimestamp(&metav1.Time{Time: time.Now()})
 }
+
+func WithConfigmapOwnerRef(ownerref *metav1.OwnerReference) reconcilertesting.ConfigMapOption {
+	return func(cg *corev1.ConfigMap) {
+		cg.ObjectMeta.OwnerReferences = []metav1.OwnerReference{*ownerref}
+	}
+}