Repository: knative-sandbox/eventing-kafka-broker
Pull request #2817 (2022-12-01 07:20:14)

Missing unit tests for namespaced broker
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Missing unit tests for namespaced broker
- Didn't add all cases, as we're wrapping existing well-tested reconciler
-

<!--
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.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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/2817#" title="Author self-approved">aliok</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)~~ [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: /cherry-pick release-1.8
/cherry-pick release-1.7
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/2817#issuecomment-1333318536):

>/cherry-pick release-1.8
>/cherry-pick release-1.7


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

<details>

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

>/cherry-pick release-1.7


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>
aliok: /assign matzew
/assign pierdipi
/assign creydr 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2817](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (f146086) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/2ae4aa33329b13b763a8de5d7559da95d725b7ad?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2ae4aa3) will **decrease** coverage by `3.36%`.
> The diff coverage is `n/a`.

```diff
@@             Coverage Diff              @@
##               main    #2817      +/-   ##
============================================
- Coverage     61.58%   58.22%   -3.37%     
============================================
  Files           153       76      -77     
  Lines         10767     8148    -2619     
  Branches        231        0     -231     
============================================
- Hits           6631     4744    -1887     
+ Misses         3668     3048     -620     
+ Partials        468      356     -112     
```

| 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/2817?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...venting/kafka/broker/receiver/IngressProducer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9JbmdyZXNzUHJvZHVjZXIuamF2YQ==) | | |
| [...a/broker/dispatcher/impl/RecordDispatcherImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9SZWNvcmREaXNwYXRjaGVySW1wbC5qYXZh) | | |
| [...atcher/impl/filter/subscriptionsapi/AnyFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9BbnlGaWx0ZXIuamF2YQ==) | | |
| [.../kafka/broker/core/eventbus/ContractPublisher.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RQdWJsaXNoZXIuamF2YQ==) | | |
| [...venting/kafka/broker/core/tracing/TracingSpan.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdHJhY2luZy9UcmFjaW5nU3Bhbi5qYXZh) | | |
| [.../broker/dispatcher/impl/ConsumerRecordContext.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9Db25zdW1lclJlY29yZENvbnRleHQuamF2YQ==) | | |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | | |
| [...ve/eventing/kafka/broker/core/metrics/Metrics.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvbWV0cmljcy9NZXRyaWNzLmphdmE=) | | |
| [...eventing/kafka/broker/receiver/RequestContext.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZXF1ZXN0Q29udGV4dC5qYXZh) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| ... and [69 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2817/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

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]: @aliok: 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
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 15387e3e7d72d1e102598f6d57a6a7a7d961b095 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2817/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1598219217721626624) | true | `/test channel-integration-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 -->
aliok: /test channel-integration-tests-sasl-plain
knative-prow[bot]: @aliok: 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
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 15387e3e7d72d1e102598f6d57a6a7a7d961b095 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2817/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1598231387570180096) | true | `/test channel-integration-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 -->
aliok: /test channel-integration-tests-sasl-plain

```
Could not resolve dependencies for project dev.knative.eventing.kafka.broker:core:jar:1.0-SNAPSHOT: The following artifacts could not be resolved: dev.knative.eventing.kafka.broker:contract:jar:1.0-SNAPSHOT, io.vertx:vertx-kafka-client:jar:4.3.2: Could not find artifact dev.knative.eventing.kafka.broker:contract:jar:1.0-SNAPSHOT in sonatype
```
knative-prow[bot]: @aliok: 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
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 15387e3e7d72d1e102598f6d57a6a7a7d961b095 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2817/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1598236054941339648) | true | `/test channel-integration-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 -->
aliok: /test channel-integration-tests-sasl-plain

```
Waiting for first events for foo69 for 61s
takes too long to receive events
```

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 a78c282d0..07d626cba 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 b2a6d4344..16864013a 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 8194717e4..2af9d7705 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -219,6 +219,12 @@ func WithAutoscalingAnnotationsTrigger() reconcilertesting.TriggerOption {
 
 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{
@@ -299,17 +305,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 8483db71b..e354b2de5 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"
@@ -229,3 +231,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}
+	}
+}
diff --git a/test/config/sacura/resources/200-config.yaml b/test/config/sacura/resources/200-config.yaml
index f5d260761..f4ecc6f53 100644
--- a/test/config/sacura/resources/200-config.yaml
+++ b/test/config/sacura/resources/200-config.yaml
@@ -13,5 +13,6 @@ data:
     receiver:
       port: 8080
       timeout: 5m
+      maxDuplicatesPercentage: 0.1
     duration: 10m
     timeout: 5m