Repository: knative-sandbox/eventing-kafka-broker
Pull request #1545 (2021-11-25 13:31:00)

[main] Upgrade to latest dependencies
Cron -knative-prow-robot

/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

Produced by: knative-sandbox/knobots/actions/update-deps

Details:
```
bumping knative.dev/eventing-kafka b50f5d4...4089422:
  > 4089422 [KafkaChannels] Pass-through "additional headers" (# 1001)
  > 10f1a91 upgrade to latest dependencies (# 1002)
  > eaa0e79 Update actions (# 1004)
  > e9e7ff8 upgrade to latest dependencies (# 1000)
bumping knative.dev/eventing 820db20...e73a69b:
  > e73a69b upgrade to latest dependencies (# 5941)
  > 7fc053b upgrade to latest dependencies (# 5931)
bumping knative.dev/pkg 787aec5...438776b:
  > 438776b Drop unused test/cmd package (# 2366)
  > 754f332 Drop indirect dependencies from go.mod (# 2359)
  > 7f59158 Add ConfigMap parsing for int16 (# 2365)
  > 608fc87 Revert "Drop unnecessary list creation from bucketer (# 2360)" (# 2361)
  > 6352c0c Drop unnecessary list creation from bucketer (# 2360)
```

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1545#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **dprotaso** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @dprotaso` in a comment when ready.

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/main/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":["dprotaso"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1545](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0ed5d89) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/1f5f46f8ff2834f84e37cbe7499b971517ed7993?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1f5f46f) will **decrease** coverage by `5.72%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1545      +/-   ##
============================================
- Coverage     75.29%   69.56%   -5.73%     
============================================
  Files           107       41      -66     
  Lines          4128     2228    -1900     
  Branches        165        0     -165     
============================================
- Hits           3108     1550    -1558     
+ Misses          765      509     -256     
+ Partials        255      169      -86     
```

| 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/1545?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...fka/broker/core/eventbus/ContractMessageCodec.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvZXZlbnRidXMvQ29udHJhY3RNZXNzYWdlQ29kZWMuamF2YQ==) | | |
| [...broker/dispatcher/impl/consumer/OffsetManager.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PZmZzZXRNYW5hZ2VyLmphdmE=) | | |
| [...enting/kafka/broker/core/utils/Configurations.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29uZmlndXJhdGlvbnMuamF2YQ==) | | |
| [...receiver/impl/handler/MethodNotAllowedHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvTWV0aG9kTm90QWxsb3dlZEhhbmRsZXIuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| [...cher/impl/consumer/CloudEventOverridesMutator.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50T3ZlcnJpZGVzTXV0YXRvci5qYXZh) | | |
| [...ker/receiver/impl/StrictRequestToRecordMapper.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1N0cmljdFJlcXVlc3RUb1JlY29yZE1hcHBlci5qYXZh) | | |
| [...spatcher/impl/consumer/CloudEventDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9DbG91ZEV2ZW50RGVzZXJpYWxpemVyLmphdmE=) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [...ka/broker/core/security/KubernetesCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvS3ViZXJuZXRlc0NyZWRlbnRpYWxzLmphdmE=) | | |
| ... and [56 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1f5f46f...0ed5d89](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1545?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @knative-automation: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 0ed5d897217835463617a7fce87dc972ea8f484b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1545/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1463862795106783232) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1545). [Your PR dashboard](https://gubernator.knative.dev/pr/knative-automation).

<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: /hold I don't want to rebase again the https://github.com/knative-sandbox/eventing-kafka-broker/pull/617 PR, :)
knative-prow-robot: @knative-automation: PR needs rebase.

<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.
</details>
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1545#" title="Author self-approved">knative-automation</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1545#pullrequestreview-819025933" title="Approved">pierDipi</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)~~ [pierDipi]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/control-plane/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 5d1ce3585..272a9cd36 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -57,7 +57,7 @@ func main() {
 
 		// Broker controller
 		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return broker.NewController(ctx, watcher, brokerEnv, "")
+			return broker.NewController(ctx, watcher, brokerEnv)
 		},
 
 		// Trigger controller
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 3aaae1d0d..d9108de63 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -31,7 +31,9 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	eventingcorev1beta1 "knative.dev/eventing/pkg/apis/eventing/v1"
 
+	eventingv1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1"
 	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 )
 
@@ -44,7 +46,9 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"): &sourcesv1beta1.KafkaSource{},
 }
 
-var callbacks = map[schema.GroupVersionKind]validation.Callback{}
+var callbacks = map[schema.GroupVersionKind]validation.Callback{
+	eventingcorev1beta1.SchemeGroupVersion.WithKind("Broker"): eventingv1.BrokerValidationCallback(),
+}
 
 func NewDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 
diff --git a/control-plane/config/200-webhook/100-webhook-cluster-role.yaml b/control-plane/config/200-webhook/100-webhook-cluster-role.yaml
index 6edd095cd..830aa628f 100644
--- a/control-plane/config/200-webhook/100-webhook-cluster-role.yaml
+++ b/control-plane/config/200-webhook/100-webhook-cluster-role.yaml
@@ -90,3 +90,11 @@ rules:
     verbs:
       - "update"
 
+  # Eventing resources care about
+  - apiGroups:
+      - "eventing.knative.dev"
+    resources:
+      - "brokers"
+    verbs:
+      - list
+      - get
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation.go b/control-plane/pkg/apis/eventing/v1/broker_validation.go
new file mode 100644
index 000000000..d8e0f11dd
--- /dev/null
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation.go
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2021 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 v1
+
+import (
+	"context"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/webhook"
+	"knative.dev/pkg/webhook/resourcesemantics/validation"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+)
+
+type BrokerPartial struct {
+	// +optional
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+	// Spec defines the desired state of the Broker.
+	Spec BrokerPartialSpec `json:"spec,omitempty"`
+}
+
+type BrokerPartialSpec struct {
+	// Config is a KReference to the configuration that specifies
+	// configuration options for this Broker. For example, this could be
+	// a pointer to a ConfigMap.
+	// +optional
+	Config *duckv1.KReference `json:"config,omitempty"`
+}
+
+func (b BrokerPartialSpec) Validate(context.Context) error {
+	if b.Config == nil {
+		return apis.ErrMissingField("config").ViaField("spec")
+	}
+	if strings.ToLower(b.Config.Kind) != "configmap" {
+		return apis.ErrInvalidValue(b.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
+	}
+	return nil
+}
+
+func validateBrokerFromUnstructured(ctx context.Context, unstructured *unstructured.Unstructured) error {
+	broker := BrokerPartial{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &broker); err != nil {
+		return err
+	}
+	if class, ok := broker.Annotations[eventing.BrokerClassAnnotationKey]; !ok || class != kafka.BrokerClass {
+		return nil
+	}
+	return broker.Spec.Validate(ctx)
+}
+
+func BrokerValidationCallback() validation.Callback {
+	return validation.NewCallback(validateBrokerFromUnstructured, webhook.Create, webhook.Update)
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index a34ea47e8..84b3458bb 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -20,9 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
-	"sync"
 
-	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -30,7 +28,6 @@ import (
 	"k8s.io/client-go/util/retry"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/pkg/controller"
-	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
 
@@ -56,11 +53,7 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	KafkaDefaultTopicDetails     sarama.TopicDetail
-	KafkaDefaultTopicDetailsLock sync.RWMutex
-	bootstrapServers             []string
-	bootstrapServersLock         sync.RWMutex
-	ConfigMapLister              corelisters.ConfigMapLister
+	ConfigMapLister corelisters.ConfigMapLister
 
 	// 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.
@@ -345,11 +338,6 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*
 
 	logger.Debug("broker config", zap.Any("broker.spec.config", broker.Spec.Config))
 
-	if broker.Spec.Config == nil {
-		tc, err := r.defaultConfig()
-		return tc, nil, err
-	}
-
 	if strings.ToLower(broker.Spec.Config.Kind) != "configmap" { // TODO: is there any constant?
 		return nil, nil, fmt.Errorf("supported config Kind: ConfigMap - got %s", broker.Spec.Config.Kind)
 	}
@@ -372,27 +360,6 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*
 	return topicConfig, cm, nil
 }
 
-func (r *Reconciler) defaultTopicDetail() sarama.TopicDetail {
-	r.KafkaDefaultTopicDetailsLock.RLock()
-	defer r.KafkaDefaultTopicDetailsLock.RUnlock()
-
-	// copy the default topic details
-	topicDetail := r.KafkaDefaultTopicDetails
-	return topicDetail
-}
-
-func (r *Reconciler) defaultConfig() (*kafka.TopicConfig, error) {
-	bootstrapServers, err := r.getDefaultBootstrapServersOrFail()
-	if err != nil {
-		return nil, err
-	}
-
-	return &kafka.TopicConfig{
-		TopicDetail:      r.defaultTopicDetail(),
-		BootstrapServers: bootstrapServers,
-	}, nil
-}
-
 func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string, broker *eventing.Broker, secret *corev1.Secret, config *kafka.TopicConfig) (*contract.Resource, error) {
 	resource := &contract.Resource{
 		Uid:    string(broker.UID),
@@ -424,61 +391,3 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 
 	return resource, nil
 }
-
-func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev1.ConfigMap) {
-
-	logger := logging.FromContext(ctx).Desugar()
-
-	return func(configMap *corev1.ConfigMap) {
-
-		topicConfig, err := kafka.TopicConfigFromConfigMap(logger, configMap)
-		if err != nil {
-			logger.Error("failed to build broker config from configmap",
-				zap.String("configMap.Namespace", configMap.Namespace),
-				zap.String("configMap.Name", configMap.Name),
-				zap.Any("configMap.Data", configMap.Data),
-				zap.Error(err))
-			return
-		}
-
-		logger.Debug("new defaults",
-			zap.Any("topicDetail", topicConfig.TopicDetail),
-			zap.String("BootstrapServers", topicConfig.GetBootstrapServers()),
-		)
-
-		r.SetDefaultTopicDetails(topicConfig.TopicDetail)
-		r.SetBootstrapServers(topicConfig.GetBootstrapServers())
-	}
-}
-
-// SetBootstrapServers change kafka bootstrap brokers addresses.
-// servers: a comma separated list of brokers to connect to.
-func (r *Reconciler) SetBootstrapServers(servers string) {
-	if servers == "" {
-		return
-	}
-
-	addrs := kafka.BootstrapServersArray(servers)
-
-	r.bootstrapServersLock.Lock()
-	r.bootstrapServers = addrs
-	r.bootstrapServersLock.Unlock()
-}
-
-func (r *Reconciler) SetDefaultTopicDetails(topicDetail sarama.TopicDetail) {
-	r.KafkaDefaultTopicDetailsLock.Lock()
-	defer r.KafkaDefaultTopicDetailsLock.Unlock()
-
-	r.KafkaDefaultTopicDetails = topicDetail
-}
-
-func (r *Reconciler) getDefaultBootstrapServersOrFail() ([]string, error) {
-	r.bootstrapServersLock.RLock()
-	defer r.bootstrapServersLock.RUnlock()
-
-	if len(r.bootstrapServers) == 0 {
-		return nil, fmt.Errorf("no %s provided", kafka.BootstrapServersConfigMapKey)
-	}
-
-	return r.bootstrapServers, nil
-}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 6973cf631..798d178b2 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -19,13 +19,11 @@ package broker
 import (
 	"context"
 	"errors"
-	"fmt"
 	"net/http"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
@@ -48,11 +46,11 @@ import (
 )
 
 const (
-	DefaultNumPartitions     = 10
-	DefaultReplicationFactor = 1
+	DefaultNumPartitions     = 20
+	DefaultReplicationFactor = 5
 )
 
-func NewController(ctx context.Context, watcher configmap.Watcher, env *config.Env, bootstrapServers string) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher, env *config.Env) *controller.Impl {
 
 	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
@@ -71,12 +69,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
-		KafkaDefaultTopicDetails: sarama.TopicDetail{
-			NumPartitions:     DefaultNumPartitions,
-			ReplicationFactor: DefaultReplicationFactor,
-		},
-		ConfigMapLister: configmapInformer.Lister(),
-		Env:             env,
+		ConfigMapLister:            configmapInformer.Lister(),
+		Env:                        env,
 	}
 
 	logger := logging.FromContext(ctx)
@@ -89,10 +83,6 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 		)
 	}
 
-	if bootstrapServers != "" {
-		reconciler.SetBootstrapServers(bootstrapServers)
-	}
-
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
 
 	reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
@@ -143,15 +133,6 @@ func NewController(ctx context.Context, watcher configmap.Watcher, env *config.E
 		},
 	})
 
-	cm, err := reconciler.KubeClient.CoreV1().ConfigMaps(env.SystemNamespace).Get(ctx, env.GeneralConfigMapName, metav1.GetOptions{})
-	if err != nil {
-		panic(fmt.Errorf("failed to get config map %s/%s: %w", env.SystemNamespace, env.GeneralConfigMapName, err))
-	}
-
-	reconciler.ConfigMapUpdated(ctx)(cm)
-
-	watcher.Watch(env.GeneralConfigMapName, reconciler.ConfigMapUpdated(ctx))
-
 	return impl
 }
 
diff --git a/go.mod b/go.mod
index 8888dd687..8ee6d5131 100644
--- a/go.mod
+++ b/go.mod
@@ -25,9 +25,9 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2
-	knative.dev/eventing-kafka v0.27.1-0.20211123070449-b50f5d4e986f
+	knative.dev/eventing v0.27.1-0.20211130054408-e73a69b875eb
+	knative.dev/eventing-kafka v0.27.1-0.20211130062708-40894229b566
 	knative.dev/hack v0.0.0-20211122162614-813559cefdda
-	knative.dev/pkg v0.0.0-20211123135150-787aec59e70a
+	knative.dev/pkg v0.0.0-20211129195804-438776b3c87c
 	knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f
 )
diff --git a/go.sum b/go.sum
index 0d51d9479..c8952e45a 100644
--- a/go.sum
+++ b/go.sum
@@ -1331,23 +1331,21 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211115173555-0910714d1183/go.mod h1:h/Ac2W8Db0pgTo4baEfzRhn+MPpxJnkSQTBTZjthwXY=
-knative.dev/eventing v0.27.1-0.20211122192117-85f324cc2df3/go.mod h1:Iuc/ZN7I48IFylgcD4KLjkj8VwajmbkDz+bH1dfAhaQ=
-knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2 h1:QwyD6phsrqx5hDplMkFZ6NJPLMSnxN3YjAw+Qnnrbpg=
-knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2/go.mod h1:fnBpTKtPOxC7WOaQY6WUiUAVBXTb44njuhTi7LXHGVc=
-knative.dev/eventing-kafka v0.27.1-0.20211123070449-b50f5d4e986f h1:vr9uJLxZgomJh0XCgDFwKgEasmJLAcHZ/MBLFIpuMOk=
-knative.dev/eventing-kafka v0.27.1-0.20211123070449-b50f5d4e986f/go.mod h1:oVSwyK9HILGLY2wkwIXQ4/iE2GTbfgCwD9QFXJzpO0w=
+knative.dev/eventing v0.27.1-0.20211126120551-7fc053b79089/go.mod h1:gGQdsi8sNQKIfCFJ58CngPFBrnO5Oqn8pNjZNFMDg88=
+knative.dev/eventing v0.27.1-0.20211130054408-e73a69b875eb h1:C1neO/Z3+oVaPXA+wKXUvwQdYMlTmEN2kHWkfzRi1tM=
+knative.dev/eventing v0.27.1-0.20211130054408-e73a69b875eb/go.mod h1:UEnzKapiHxRsFUF8wbJXFMsfktaEtGltBZdTebSNRCU=
+knative.dev/eventing-kafka v0.27.1-0.20211130062708-40894229b566 h1:XpxiBh8tvCd1vUlAr9juuhomkFpPhrLgS5YEthmNYVE=
+knative.dev/eventing-kafka v0.27.1-0.20211130062708-40894229b566/go.mod h1:VlpcJySuKNP4Xx9nX5p2Gf2b4f+FuvqrESg5njGP/Cs=
 knative.dev/hack v0.0.0-20211101195839-11d193bf617b/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211112192837-128cf0150a69/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack v0.0.0-20211117134436-69a2295d54ce/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211122162614-813559cefdda h1:WBfGcmu5pN+RCxDotKVYtWlJVI2V4Hvah+XD9mcSR/o=
 knative.dev/hack v0.0.0-20211122162614-813559cefdda/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211117134436-69a2295d54ce/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/hack/schema v0.0.0-20211122162614-813559cefdda/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211101212339-96c0204a70dc/go.mod h1:SkfDk9bWIiNZD7XtILGkG7AKVyF/M6M0bGxLgl0SYL8=
 knative.dev/pkg v0.0.0-20211115071955-517ef0292b53/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211120133512-d016976f2567/go.mod h1:VqUp1KWJqpTDNoiSI/heaX3uMdubImslJE2tBkP+Bbw=
-knative.dev/pkg v0.0.0-20211123135150-787aec59e70a h1:mgE9GdkjfPJc7KpgBox52dSy1GsQmsFTWXy+1HjEn9I=
-knative.dev/pkg v0.0.0-20211123135150-787aec59e70a/go.mod h1:fZUlVceKtVNyFU6LokWGk2a9QFQXeSTtnbKOjjkQ690=
+knative.dev/pkg v0.0.0-20211125172117-608fc877e946/go.mod h1:fZUlVceKtVNyFU6LokWGk2a9QFQXeSTtnbKOjjkQ690=
+knative.dev/pkg v0.0.0-20211129195804-438776b3c87c h1:TcVqa/i/E/KcusxjroLEaMB2bDV42lYzN2Z+4soudE8=
+knative.dev/pkg v0.0.0-20211129195804-438776b3c87c/go.mod h1:AKPae1Cmj+k0GWXWnF2tKY7q5qPa1mTD7oCP4OeMvEM=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f h1:bSdsp5av09LqSF4iXo8LCCDEENJ+JEbzzdgHBoJdf2w=
 knative.dev/reconciler-test v0.0.0-20211112132636-ae9e2e21972f/go.mod h1:gTsbLk496j/M9xqMpx/liyCQ0X3bwDpRtcs2Zzws364=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
diff --git a/vendor/knative.dev/pkg/configmap/parse.go b/vendor/knative.dev/pkg/configmap/parse.go
index faeacab14..e93f9af2a 100644
--- a/vendor/knative.dev/pkg/configmap/parse.go
+++ b/vendor/knative.dev/pkg/configmap/parse.go
@@ -53,6 +53,20 @@ func AsBool(key string, target *bool) ParseFunc {
 	}
 }
 
+// AsInt16 parses the value at key as an int16 into the target, if it exists.
+func AsInt16(key string, target *int16) ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			val, err := strconv.ParseInt(raw, 10, 16)
+			if err != nil {
+				return fmt.Errorf("failed to parse %q: %w", key, err)
+			}
+			*target = int16(val)
+		}
+		return nil
+	}
+}
+
 // AsInt32 parses the value at key as an int32 into the target, if it exists.
 func AsInt32(key string, target *int32) ParseFunc {
 	return func(data map[string]string) error {
@@ -95,6 +109,20 @@ func AsInt(key string, target *int) ParseFunc {
 	}
 }
 
+// AsUint16 parses the value at key as an uint16 into the target, if it exists.
+func AsUint16(key string, target *uint16) ParseFunc {
+	return func(data map[string]string) error {
+		if raw, ok := data[key]; ok {
+			val, err := strconv.ParseUint(raw, 10, 16)
+			if err != nil {
+				return fmt.Errorf("failed to parse %q: %w", key, err)
+			}
+			*target = uint16(val)
+		}
+		return nil
+	}
+}
+
 // AsUint32 parses the value at key as an uint32 into the target, if it exists.
 func AsUint32(key string, target *uint32) ParseFunc {
 	return func(data map[string]string) error {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 6f2122b9b..043438779 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1112,7 +1112,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.27.1-0.20211123205351-820db20be4b2
+# knative.dev/eventing v0.27.1-0.20211130054408-e73a69b875eb
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1234,7 +1234,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.27.1-0.20211123070449-b50f5d4e986f
+# knative.dev/eventing-kafka v0.27.1-0.20211130062708-40894229b566
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1294,7 +1294,7 @@ knative.dev/eventing-kafka/test/upgrade/continual
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20211123135150-787aec59e70a
+# knative.dev/pkg v0.0.0-20211129195804-438776b3c87c
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
new file mode 100644
index 000000000..0c08ba763
--- /dev/null
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2021 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 v1
+
+import (
+	"context"
+	"testing"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+func TestValidateBrokerFromUnstructured(t *testing.T) {
+	tests := []struct {
+		name         string
+		ctx          context.Context
+		unstructured *unstructured.Unstructured
+		wantErr      bool
+	}{
+		{
+			name: "no kafka broker",
+			ctx:  context.Background(),
+			unstructured: &unstructured.Unstructured{
+				Object: map[string]interface{}{"spec": map[string]interface{}{}},
+			},
+			wantErr: false,
+		},
+		{
+			name: "missing config",
+			ctx:  context.Background(),
+			unstructured: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"metadata": map[string]interface{}{
+						"annotations": map[string]string{
+							"eventing.knative.dev/broker.class": "Kafka",
+						},
+					},
+					"spec": map[string]interface{}{},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "unknown kind",
+			ctx:  context.Background(),
+			unstructured: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"metadata": map[string]interface{}{
+						"annotations": map[string]string{
+							"eventing.knative.dev/broker.class": "Kafka",
+						},
+					},
+					"spec": map[string]interface{}{
+						"config": map[string]string{
+							"apiVersion": "eventing.knative.dev/v1",
+							"kind":       "Broker",
+							"namespace":  "ns",
+							"name":       "name",
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "ConfigMap in config",
+			ctx:  context.Background(),
+			unstructured: &unstructured.Unstructured{
+				Object: map[string]interface{}{
+					"metadata": map[string]interface{}{
+						"annotations": map[string]string{
+							"eventing.knative.dev/broker.class": "Kafka",
+						},
+					},
+					"spec": map[string]interface{}{
+						"config": map[string]string{
+							"apiVersion": "v1",
+							"kind":       "ConfigMap",
+							"namespace":  "ns",
+							"name":       "name",
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if err := validateBrokerFromUnstructured(tt.ctx, tt.unstructured); (err != nil) != tt.wantErr {
+				t.Errorf("validateBrokerFromUnstructured() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index a9ea14947..a655eb2eb 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -20,9 +20,7 @@ import (
 	"context"
 	"fmt"
 	"net/url"
-	"sync"
 	"testing"
-	"time"
 
 	"k8s.io/utils/pointer"
 	"knative.dev/pkg/network"
@@ -34,7 +32,6 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/Shopify/sarama"
-	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -121,6 +118,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
@@ -173,14 +171,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled failed - probe " + prober.StatusNotReady.String(),
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
@@ -233,14 +229,14 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-				testProber:                         probertesting.MockProber(prober.StatusNotReady),
+				testProber: probertesting.MockProber(prober.StatusNotReady),
 			},
 		},
 		{
 			Name: "Reconciled failed - probe " + prober.StatusUnknown.String(),
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapWithBinaryData(&env, nil),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{
@@ -293,8 +289,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-				testProber:                         probertesting.MockProber(prober.StatusUnknown),
+				testProber: probertesting.MockProber(prober.StatusUnknown),
 			},
 		},
 		{
@@ -303,6 +298,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -352,9 +348,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - with DLS - no DLS ref namespace",
@@ -362,6 +355,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(WithNoDeadLetterSinkNamespace),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -411,14 +405,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Failed to create topic",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				BrokerReceiverPod(env.SystemNamespace, nil),
 				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
@@ -447,8 +439,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 			OtherTestData: map[string]interface{}{
-				wantErrorOnCreateTopic:             createTopicError,
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
+				wantErrorOnCreateTopic: createTopicError,
 			},
 		},
 		{
@@ -457,6 +448,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
@@ -513,14 +505,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - config map not readable",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapWithBinaryData(&env, []byte(`{"hello": "world"}`)),
 				NewService(),
 				BrokerReceiverPod(env.SystemNamespace, nil),
@@ -567,14 +557,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - preserve config map previous state",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -648,9 +636,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - update existing broker while preserving others",
@@ -668,6 +653,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						}
 					},
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -746,9 +732,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - remove existing broker DLS while preserving others",
@@ -758,6 +741,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						broker.Spec.Delivery = &eventingduck.DeliverySpec{}
 					},
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -830,14 +814,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - increment volume generation",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -907,9 +889,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Failed to resolve DLS",
@@ -921,6 +900,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						}
 					},
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -974,14 +954,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "No bootstrap.servers provided",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig("", 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 1,
@@ -996,7 +974,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
-					"failed to get contract configuration: no bootstrap.servers provided",
+					"failed to get contract configuration: unable to build topic config from configmap: error validating topic config from configmap invalid configuration - numPartitions: 20 - replicationFactor: 5 - bootstrapServers: [] - ConfigMap data: map[bootstrap.servers: default.topic.partitions:20 default.topic.replication.factor:5] - ConfigMap data: map[bootstrap.servers: default.topic.partitions:20 default.topic.replication.factor:5]",
 				),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
@@ -1007,7 +985,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
 						StatusBrokerDataPlaneAvailable,
-						StatusBrokerConfigNotParsed("no bootstrap.servers provided"),
+						StatusBrokerConfigNotParsed("unable to build topic config from configmap: error validating topic config from configmap invalid configuration - numPartitions: 20 - replicationFactor: 5 - bootstrapServers: [] - ConfigMap data: map[bootstrap.servers: default.topic.partitions:20 default.topic.replication.factor:5] - ConfigMap data: map[bootstrap.servers: default.topic.partitions:20 default.topic.replication.factor:5]"),
 					),
 				},
 			},
@@ -1077,7 +1055,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
 				ExpectedTopicDetail: sarama.TopicDetail{
 					NumPartitions:     20,
 					ReplicationFactor: 5,
@@ -1156,7 +1133,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				},
 			},
 			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
 				ExpectedTopicDetail: sarama.TopicDetail{
 					NumPartitions:     20,
 					ReplicationFactor: 5,
@@ -1199,9 +1175,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Unsupported Kind as config",
@@ -1255,9 +1228,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - keep all existing triggers",
@@ -1265,6 +1235,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1368,14 +1339,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "no data plane pods running",
 			Objects: []runtime.Object{
 				NewBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -1406,6 +1375,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					WithDelivery(),
 					WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -1461,9 +1431,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - with retry config - linear",
@@ -1472,6 +1439,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					WithDelivery(),
 					WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -1527,9 +1495,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - with no retry num",
@@ -1538,6 +1503,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					WithDelivery(),
 					WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -1590,9 +1556,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - with retry config - no retry delay - use default delay",
@@ -1601,6 +1564,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					WithDelivery(),
 					WithRetry(pointer.Int32Ptr(10), &linear, nil),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
 				}, &env),
@@ -1656,9 +1620,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - unchanged",
@@ -1666,6 +1627,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1705,9 +1667,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - unchanged contract - changed data plane pods annotation",
@@ -1715,6 +1674,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 				NewBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1757,9 +1717,6 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 					),
 				},
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 	}
 
@@ -1789,6 +1746,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1807,9 +1765,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 2,
 				}),
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - with DLS",
@@ -1817,6 +1772,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1835,14 +1791,12 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 2,
 				}),
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Failed to delete topic",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1870,8 +1824,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}),
 			},
 			OtherTestData: map[string]interface{}{
-				wantErrorOnDeleteTopic:             deleteTopicError,
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
+				wantErrorOnDeleteTopic: deleteTopicError,
 			},
 		},
 		{
@@ -1880,6 +1833,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewDeletedBroker(
 					WithDelivery(),
 				),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewService(),
 			},
 			Key: testKey,
@@ -1887,14 +1841,12 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - preserve config map previous state",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1924,14 +1876,12 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 					Generation: 6,
 				}),
 			},
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 		{
 			Name: "Reconciled normal - topic doesn't exist",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1962,14 +1912,14 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}),
 			},
 			OtherTestData: map[string]interface{}{
-				wantErrorOnDeleteTopic:             sarama.ErrUnknownTopicOrPartition,
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
+				wantErrorOnDeleteTopic: sarama.ErrUnknownTopicOrPartition,
 			},
 		},
 		{
 			Name: "Reconciled normal - no broker found in config map",
 			Objects: []runtime.Object{
 				NewDeletedBroker(),
+				BrokerConfig(bootstrapServers, 20, 5),
 				NewConfigMapFromContract(&contract.Contract{
 					Resources: []*contract.Resource{
 						{
@@ -1982,9 +1932,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 				}, &env),
 			},
 			Key: testKey,
-			OtherTestData: map[string]interface{}{
-				kafka.BootstrapServersConfigMapKey: bootstrapServers,
-			},
 		},
 	}
 
@@ -1997,8 +1944,6 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 
 func useTable(t *testing.T, table TableTest, env *config.Env) {
 
-	testCtx, cancel := context.WithCancel(context.Background())
-
 	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		defaultTopicDetail := sarama.TopicDetail{
@@ -2016,11 +1961,6 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			onDeleteTopicError = want.(error)
 		}
 
-		bootstrapServers := ""
-		if bs, ok := row.OtherTestData[kafka.BootstrapServersConfigMapKey]; ok {
-			bootstrapServers = bs.(string)
-		}
-
 		expectedTopicDetail := defaultTopicDetail
 		if td, ok := row.OtherTestData[ExpectedTopicDetail]; ok {
 			expectedTopicDetail = td.(sarama.TopicDetail)
@@ -2043,9 +1983,7 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
 			},
-			KafkaDefaultTopicDetails:     defaultTopicDetail,
-			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
-			ConfigMapLister:              listers.GetConfigMapLister(),
+			ConfigMapLister: listers.GetConfigMapLister(),
 			NewKafkaClusterAdminClient: func(_ []string, _ *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
 					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
@@ -2059,7 +1997,6 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 			Prober:      proberMock,
 			IngressHost: network.GetServiceHostname(env.IngressName, env.SystemNamespace),
 		}
-		reconciler.SetBootstrapServers(bootstrapServers)
 
 		reconciler.ConfigMapTracker = &FakeTracker{}
 		reconciler.SecretTracker = &FakeTracker{}
@@ -2076,51 +2013,8 @@ func useTable(t *testing.T, table TableTest, env *config.Env) {
 
 		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))
 
-		// periodically update default topic details to simulate concurrency.
-		go func() {
-
-			ticker := time.NewTicker(10 * time.Millisecond)
-
-			for {
-				select {
-				case <-testCtx.Done():
-					return
-				case <-ticker.C:
-					reconciler.SetDefaultTopicDetails(defaultTopicDetail)
-				}
-			}
-		}()
-
 		return r
 	}))
-
-	cancel()
-}
-
-func TestConfigMapUpdate(t *testing.T) {
-
-	cm := corev1.ConfigMap{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "cmname",
-			Namespace: "cmnamespace",
-		},
-		Data: map[string]string{
-			kafka.DefaultTopicNumPartitionConfigMapKey:      "42",
-			kafka.DefaultTopicReplicationFactorConfigMapKey: "3",
-			kafka.BootstrapServersConfigMapKey:              "server1,server2",
-		},
-	}
-
-	reconciler := Reconciler{}
-
-	ctx, _ := SetupFakeContext(t)
-
-	reconciler.ConfigMapUpdated(ctx)(&cm)
-
-	assert.Equal(t, reconciler.KafkaDefaultTopicDetails, sarama.TopicDetail{
-		NumPartitions:     42,
-		ReplicationFactor: 3,
-	})
 }
 
 func patchFinalizers() clientgotesting.PatchActionImpl {
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index d33eb2e1d..9fd2aa1a1 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -68,7 +68,6 @@ func TestNewController(t *testing.T) {
 			},
 		}),
 		env,
-		"",
 	)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index cca4cf32f..df1c10289 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -59,6 +59,9 @@ func NewBroker(options ...reconcilertesting.BrokerOption) runtime.Object {
 		append(
 			[]reconcilertesting.BrokerOption{
 				reconcilertesting.WithBrokerClass(kafka.BrokerClass),
+				WithBrokerConfig(
+					KReference(BrokerConfig("", 20, 5)),
+				),
 				func(broker *eventing.Broker) {
 					broker.UID = BrokerUUID
 				},