Repository: knative-sandbox/eventing-kafka-broker
Pull request #64 (2020-07-28 07:28:53)

Remove data plane config map from artifacts and create it if not found
If users apply the release artifact, they will create the data plane `ConfigMap` with no data in it. After creating objects (Brokers and Triggers) re-applying the release artifact will lead to a data loss because the `ConfigMap` will be re-created with no data in it. This PR fixes this behavior by using the control plane for creating such `ConfigMap` if it is not there.

## Proposed Changes

- Create data plane config map if it's not found

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/64#" title="Author self-approved">pierDipi</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mattmoor**
You can assign the PR to them by writing `/assign @mattmoor` 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/master/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":["mattmoor"]} -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288013631203577856/artifacts/line-cov.html#file0) | 90.8% | 89.2% | -1.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288013631203577856/artifacts/line-cov.html#file1) | 85.7% | 84.2% | -1.5
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288013631203577856/artifacts/line-cov.html#file2) | 85.4% | 84.3% | -1.1

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/64#" title="Author self-approved">pierDipi</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/64#pullrequestreview-456396907" title="Approved">slinkydeveloper</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/master/OWNERS)~~ [slinkydeveloper]

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":[]} -->
knative-prow-robot: @pierDipi: 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>
slinkydeveloper: /lgtm
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288024829596798979/artifacts/line-cov.html#file0) | 90.8% | 89.2% | -1.7
[control-plane/pkg/reconciler/broker/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288024829596798979/artifacts/line-cov.html#file1) | 85.7% | 84.2% | -1.5
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/64/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1288024829596798979/artifacts/line-cov.html#file2) | 85.4% | 84.3% | -1.1


Patch:
diff --git a/control-plane/config/200-controller-cluster-role.yaml b/control-plane/config/200-controller-cluster-role.yaml
index 16d18aa0e..79fb6916b 100644
--- a/control-plane/config/200-controller-cluster-role.yaml
+++ b/control-plane/config/200-controller-cluster-role.yaml
@@ -30,6 +30,7 @@ rules:
       - list
       - watch
       - update
+      - create
   - apiGroups:
       - "*"
     resources:
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index 73b19bf6b..e71176e1c 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -7,6 +7,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
@@ -47,10 +48,30 @@ type Reconciler struct {
 	SystemNamespace             string
 }
 
-func (r *Reconciler) GetDataPlaneConfigMap() (*corev1.ConfigMap, error) {
-	return r.KubeClient.CoreV1().
+func (r *Reconciler) GetOrCreateDataPlaneConfigMap() (*corev1.ConfigMap, error) {
+
+	cm, err := r.KubeClient.CoreV1().
 		ConfigMaps(r.DataPlaneConfigMapNamespace).
 		Get(r.DataPlaneConfigMapName, metav1.GetOptions{})
+
+	if apierrors.IsNotFound(err) {
+		cm, err = r.createDataPlaneConfigMap()
+	}
+
+	return cm, err
+}
+
+func (r *Reconciler) createDataPlaneConfigMap() (*corev1.ConfigMap, error) {
+	return r.KubeClient.CoreV1().ConfigMaps(r.DataPlaneConfigMapNamespace).Create(&corev1.ConfigMap{
+		TypeMeta: metav1.TypeMeta{},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      r.DataPlaneConfigMapName,
+			Namespace: r.DataPlaneConfigMapNamespace,
+		},
+		BinaryData: map[string][]byte{
+			ConfigMapDataKey: []byte(""),
+		},
+	})
 }
 
 // GetDataPlaneConfigMapData extracts brokers and triggers data from the given config map.
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index a5991c945..3c6ffb7c0 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -22,7 +22,6 @@ import (
 	"math"
 	"strings"
 	"sync"
-	"time"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -53,10 +52,16 @@ type Reconciler struct {
 
 	Resolver *resolver.URIResolver
 
-	KafkaClusterAdmin            sarama.ClusterAdmin
-	KafkaClusterAdminLock        sync.RWMutex
+	// TODO these configurations should live in each Broker configuration, so that we don't assume each
+	// 	Broker object use the same Kafka cluster
 	KafkaDefaultTopicDetails     sarama.TopicDetail
 	KafkaDefaultTopicDetailsLock sync.RWMutex
+	bootstrapServers             []string
+	bootstrapServersLock         sync.RWMutex
+
+	// NewClusterAdmin 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.
+	NewClusterAdmin func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error)
 
 	Configs *Configs
 }
@@ -88,7 +93,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	logger.Debug("Topic created", zap.Any("topic", topic))
 
 	// Get brokers and triggers config map.
-	brokersTriggersConfigMap, err := r.GetDataPlaneConfigMap()
+	brokersTriggersConfigMap, err := r.GetOrCreateDataPlaneConfigMap()
 	if err != nil {
 		return statusConditionManager.failedToGetBrokersTriggersConfigMap(err)
 	}
@@ -180,7 +185,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	logger.Debug("Finalizing broker")
 
 	// Get brokers and triggers config map.
-	brokersTriggersConfigMap, err := r.GetDataPlaneConfigMap()
+	brokersTriggersConfigMap, err := r.GetOrCreateDataPlaneConfigMap()
 	if err != nil {
 		return fmt.Errorf("failed to get brokers and triggers config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
 	}
@@ -288,38 +293,30 @@ func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev
 		topicDetail.ReplicationFactor = int16(replicationFactor)
 
 		r.SetDefaultTopicDetails(topicDetail)
-
-		if err := r.SetBootstrapServers(bootstrapServers); err != nil {
-			logger.Error("Failed to set bootstrap servers", zap.Error(err))
-		}
+		r.SetBootstrapServers(bootstrapServers)
 	}
 }
 
 // SetBootstrapServers change kafka bootstrap brokers addresses.
 // servers: a comma separated list of brokers to connect to.
-func (r *Reconciler) SetBootstrapServers(servers string) error {
+func (r *Reconciler) SetBootstrapServers(servers string) {
 	addrs := strings.Split(servers, ",")
 
+	r.bootstrapServersLock.Lock()
+	r.bootstrapServers = addrs
+	r.bootstrapServersLock.Unlock()
+}
+
+func (r *Reconciler) getKafkaClusterAdmin(bootstrapServers []string) (sarama.ClusterAdmin, error) {
 	config := sarama.NewConfig()
 	config.Version = sarama.MaxVersion
-	config.Net.KeepAlive = time.Second * 60
-	config.Metadata.RefreshFrequency = time.Minute
 
-	kafkaClusterAdmin, err := NewClusterAdmin(addrs, config)
+	kafkaClusterAdmin, err := r.NewClusterAdmin(bootstrapServers, config)
 	if err != nil {
-		return fmt.Errorf("failed to create kafka cluster admin: %w", err)
+		return nil, fmt.Errorf("failed to create cluster admin: %w", err)
 	}
 
-	r.KafkaClusterAdminLock.Lock()
-	oldKafkaClusterAdmin := r.KafkaClusterAdmin
-	r.KafkaClusterAdmin = kafkaClusterAdmin
-	r.KafkaClusterAdminLock.Unlock()
-
-	if oldKafkaClusterAdmin != nil {
-		_ = oldKafkaClusterAdmin.Close()
-	}
-
-	return nil
+	return kafkaClusterAdmin, nil
 }
 
 func (r *Reconciler) SetDefaultTopicDetails(topicDetail sarama.TopicDetail) {
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index f8672150c..3621f3f2b 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -21,9 +21,11 @@ import (
 	"fmt"
 
 	"github.com/Shopify/sarama"
+	"go.uber.org/zap"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/tools/cache"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1beta1"
+	"knative.dev/eventing/pkg/logging"
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
@@ -46,8 +48,6 @@ const (
 	DefaultReplicationFactor = 1
 )
 
-var NewClusterAdmin = sarama.NewClusterAdmin
-
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *Configs) *controller.Impl {
 
 	eventing.RegisterAlternateBrokerConditionSet(ConditionSet)
@@ -61,6 +61,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
 			SystemNamespace:             configs.SystemNamespace,
 		},
+		NewClusterAdmin: sarama.NewClusterAdmin,
 		KafkaDefaultTopicDetails: sarama.TopicDetail{
 			NumPartitions:     DefaultNumPartitions,
 			ReplicationFactor: DefaultReplicationFactor,
@@ -68,8 +69,18 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 		Configs: configs,
 	}
 
+	logger := logging.FromContext(ctx)
+
+	_, err := reconciler.GetOrCreateDataPlaneConfigMap()
+	if err != nil {
+		logger.Fatal("Failed to get or create data plane config map",
+			zap.String("configmap", configs.DataPlaneConfigMapAsString()),
+			zap.Error(err),
+		)
+	}
+
 	if configs.BootstrapServers != "" {
-		_ = reconciler.SetBootstrapServers(configs.BootstrapServers)
+		reconciler.SetBootstrapServers(configs.BootstrapServers)
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
@@ -78,6 +89,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 
 	brokerInformer := brokerinformer.Get(ctx)
 
+	logger.Info("Register event handlers")
+
 	brokerInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: kafka.BrokerClassFilter(),
 		Handler:    controller.HandleAll(impl.Enqueue),
diff --git a/control-plane/pkg/reconciler/broker/topic.go b/control-plane/pkg/reconciler/broker/topic.go
index 9c6dabe8c..88225dee4 100644
--- a/control-plane/pkg/reconciler/broker/topic.go
+++ b/control-plane/pkg/reconciler/broker/topic.go
@@ -24,13 +24,18 @@ import (
 )
 
 func (r *Reconciler) CreateTopic(broker *eventing.Broker) (string, error) {
+	r.bootstrapServersLock.RLock()
+	defer r.bootstrapServersLock.RUnlock()
+
 	topic := Topic(broker)
 	topicDetail := r.topicDetailFromBrokerConfig(broker)
 
-	r.KafkaClusterAdminLock.Lock()
-	defer r.KafkaClusterAdminLock.Unlock()
+	kafkaClusterAdmin, err := r.getKafkaClusterAdmin(r.bootstrapServers)
+	if err != nil {
+		return "", err
+	}
 
-	createTopicError := r.KafkaClusterAdmin.CreateTopic(topic, topicDetail, true)
+	createTopicError := kafkaClusterAdmin.CreateTopic(topic, topicDetail, true)
 	if err, ok := createTopicError.(*sarama.TopicError); ok && err.Err == sarama.ErrTopicAlreadyExists {
 		return topic, nil
 	}
@@ -39,13 +44,17 @@ func (r *Reconciler) CreateTopic(broker *eventing.Broker) (string, error) {
 }
 
 func (r *Reconciler) deleteTopic(broker *eventing.Broker) (string, error) {
-
-	r.KafkaClusterAdminLock.RLock()
-	defer r.KafkaClusterAdminLock.RUnlock()
+	r.bootstrapServersLock.RLock()
+	defer r.bootstrapServersLock.RUnlock()
 
 	topic := Topic(broker)
 
-	err := r.KafkaClusterAdmin.DeleteTopic(topic)
+	kafkaClusterAdmin, err := r.getKafkaClusterAdmin(r.bootstrapServers)
+	if err != nil {
+		return "", err
+	}
+
+	err = kafkaClusterAdmin.DeleteTopic(topic)
 	if sarama.ErrUnknownTopicOrPartition == err {
 		return topic, nil
 	}
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index a7d2a658d..cc52756be 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -75,7 +75,7 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, trigger *eventing.Trigger
 	}
 
 	// Get data plane config map.
-	dataPlaneConfigMap, err := r.GetDataPlaneConfigMap()
+	dataPlaneConfigMap, err := r.GetOrCreateDataPlaneConfigMap()
 	if err != nil {
 		return fmt.Errorf("failed to get data plane config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
 	}
@@ -205,7 +205,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	statusConditionManager.propagateBrokerCondition(broker)
 
 	// Get data plane config map.
-	dataPlaneConfigMap, err := r.GetDataPlaneConfigMap()
+	dataPlaneConfigMap, err := r.GetOrCreateDataPlaneConfigMap()
 	if err != nil {
 		return statusConditionManager.failedToGetDataPlaneConfigMap(err)
 	}
diff --git a/data-plane/config/100-config-brokers-triggers.yaml b/data-plane/config/100-config-brokers-triggers.yaml
deleted file mode 100644
index 842187d11..000000000
--- a/data-plane/config/100-config-brokers-triggers.yaml
+++ /dev/null
@@ -1,25 +0,0 @@
----
-
-# 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.
-
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: kafka-broker-brokers-triggers
-  namespace: knative-eventing
-  labels:
-    eventing.knative.dev/release: devel
-binaryData:
-  data: ""


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index a6fa65e10..ade96ed2a 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -72,7 +72,7 @@ var (
 	deleteTopicError = fmt.Errorf("failed to delete topic")
 )
 
-func TestBrokeReconciler(t *testing.T) {
+func TestBrokerReconciler(t *testing.T) {
 	eventing.RegisterAlternateBrokerConditionSet(ConditionSet)
 
 	t.Parallel()
@@ -227,26 +227,48 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			},
 		},
 		{
-			Name: "Failed to get config map",
+			Name: "Config map not found - create config map",
 			Objects: []runtime.Object{
-				NewBroker(),
+				NewBroker(
+					WithDelivery(),
+				),
+				NewService(),
+				NewReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				NewDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 				&corev1.ConfigMap{
 					ObjectMeta: metav1.ObjectMeta{
 						Namespace: configs.DataPlaneConfigMapNamespace,
-						Name:      configs.DataPlaneConfigMapName + "a",
+						Name:      configs.DataPlaneConfigMapName + "a", // Use a different name
 					},
 				},
 			},
-			Key:     testKey,
-			WantErr: true,
+			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
-				Eventf(
-					corev1.EventTypeWarning,
-					"InternalError",
-					"failed to get brokers and triggers config map %s: %v",
-					configs.DataPlaneConfigMapAsString(), `configmaps "knative-eventing" not found`,
-				),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMap(&configs, nil),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &coreconfig.Brokers{
+					Broker: []*coreconfig.Broker{
+						{
+							Id:             BrokerUUID,
+							Topic:          GetTopic(),
+							DeadLetterSink: "http://test-service.test-service-namespace.svc.cluster.local/",
+							Namespace:      BrokerNamespace,
+							Name:           BrokerName,
+						},
+					},
+					VolumeGeneration: 1,
+				}),
+				ReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+				DispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -254,9 +276,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewBroker(
+						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
+						ConfigMapUpdatedReady(&configs),
 						TopicReady,
-						FailedToGetConfigMap(&configs),
+						Addressable(&configs),
 					),
 				},
 			},
@@ -804,26 +828,20 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 			},
 		},
 		{
-			Name: "Failed to get config map",
+			Name: "Config map not found - create config map",
 			Objects: []runtime.Object{
-				NewDeletedBroker(),
-				&corev1.ConfigMap{
-					ObjectMeta: metav1.ObjectMeta{
-						Namespace: configs.DataPlaneConfigMapNamespace,
-						Name:      configs.DataPlaneConfigMapName + "a",
-					},
-				},
-			},
-			Key:     testKey,
-			WantErr: true,
-			WantEvents: []string{
-				Eventf(
-					corev1.EventTypeWarning,
-					"InternalError",
-					"failed to get brokers and triggers config map %s: %v",
-					configs.DataPlaneConfigMapAsString(), `configmaps "knative-eventing" not found`,
+				NewDeletedBroker(
+					WithDelivery(),
 				),
+				NewService(),
+				NewReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				NewDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConfigMap(&configs, nil),
+			},
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 		},
 		{
 			Name: "Config map not readable",
@@ -960,14 +978,6 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 			onDeleteTopicError = want.(error)
 		}
 
-		clusterAdmin := &MockKafkaClusterAdmin{
-			ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
-			ExpectedTopicDetail: defaultTopicDetail,
-			ErrorOnCreateTopic:  onCreateTopicError,
-			ErrorOnDeleteTopic:  onDeleteTopicError,
-			T:                   t,
-		}
-
 		reconciler := &Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
@@ -977,7 +987,15 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
 				SystemNamespace:             configs.SystemNamespace,
 			},
-			KafkaClusterAdmin:            clusterAdmin,
+			NewClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &MockKafkaClusterAdmin{
+					ExpectedTopicName:   fmt.Sprintf("%s%s-%s", TopicPrefix, BrokerNamespace, BrokerName),
+					ExpectedTopicDetail: defaultTopicDetail,
+					ErrorOnCreateTopic:  onCreateTopicError,
+					ErrorOnDeleteTopic:  onDeleteTopicError,
+					T:                   t,
+				}, nil
+			},
 			KafkaDefaultTopicDetails:     defaultTopicDetail,
 			KafkaDefaultTopicDetailsLock: sync.RWMutex{},
 			Configs:                      configs,
@@ -1018,10 +1036,6 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 
 func TestConfigMapUpdate(t *testing.T) {
 
-	NewClusterAdmin = func(addrs []string, conf *sarama.Config) (sarama.ClusterAdmin, error) {
-		return MockKafkaClusterAdmin{}, nil
-	}
-
 	cm := corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "cmname",
@@ -1044,7 +1058,6 @@ func TestConfigMapUpdate(t *testing.T) {
 		NumPartitions:     42,
 		ReplicationFactor: 3,
 	})
-	assert.NotNil(t, reconciler.KafkaClusterAdmin)
 }
 
 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 fe36b290b..dc299faf6 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -19,7 +19,6 @@ package broker
 import (
 	"testing"
 
-	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -37,10 +36,6 @@ import (
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	NewClusterAdmin = func(addrs []string, conf *sarama.Config) (sarama.ClusterAdmin, error) {
-		return nil, nil
-	}
-
 	configs := &Configs{
 		EnvConfigs: EnvConfigs{
 			SystemNamespace:      "cm",
diff --git a/control-plane/pkg/reconciler/broker/topic_test.go b/control-plane/pkg/reconciler/broker/topic_test.go
index effdb8e60..f63f5882b 100644
--- a/control-plane/pkg/reconciler/broker/topic_test.go
+++ b/control-plane/pkg/reconciler/broker/topic_test.go
@@ -40,14 +40,16 @@ func TestCreateTopicTopicAlreadyExists(t *testing.T) {
 	errMsg := "topic already exists"
 
 	r := broker.Reconciler{
-		KafkaClusterAdmin: reconcilertesting.MockKafkaClusterAdmin{
-			ExpectedTopicName:   topic,
-			ExpectedTopicDetail: sarama.TopicDetail{},
-			ErrorOnCreateTopic: &sarama.TopicError{
-				Err:    sarama.ErrTopicAlreadyExists,
-				ErrMsg: &errMsg,
-			},
-			T: t,
+		NewClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+			return reconcilertesting.MockKafkaClusterAdmin{
+				ExpectedTopicName:   topic,
+				ExpectedTopicDetail: sarama.TopicDetail{},
+				ErrorOnCreateTopic: &sarama.TopicError{
+					Err:    sarama.ErrTopicAlreadyExists,
+					ErrMsg: &errMsg,
+				},
+				T: t,
+			}, nil
 		},
 	}
 
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 7c9fc1db7..90dac4c60 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -46,8 +46,8 @@ const (
 var DefaultConfigs = &broker.Configs{
 
 	EnvConfigs: broker.EnvConfigs{
-		DataPlaneConfigMapNamespace: "kafka-broker-brokers-triggers",
-		DataPlaneConfigMapName:      "knative-eventing",
+		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+		DataPlaneConfigMapNamespace: "knative-eventing",
 		BrokerIngressName:           "kafka-broker-receiver",
 		SystemNamespace:             "knative-eventing",
 		DataPlaneConfigFormat:       base.Json,
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 863781409..f0a696cc6 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -244,7 +244,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 			},
 		},
 		{
-			Name: "Data plane config map not found",
+			Name: "Config map not found - broker not found in config map",
 			Objects: []runtime.Object{
 				NewBroker(
 					BrokerReady,
@@ -257,13 +257,17 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 			},
 			Key:     testKey,
 			WantErr: true,
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantCreates: []runtime.Object{
+				NewConfigMap(&configs, nil),
+			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
 					fmt.Sprintf(
-						`Failed to get data plane config map %s: configmaps "knative-eventing" not found`,
+						"broker not found in data plane config map %s",
 						configs.DataPlaneConfigMapAsString(),
 					),
 				),
@@ -272,10 +276,9 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				{
 					Object: newTrigger(
 						reconcilertesting.WithInitTriggerConditions,
-						reconcilertesting.WithTriggerBrokerReady(),
-						reconcilertesting.WithTriggerDependencyFailed(
-							fmt.Sprintf("Failed to get data plane config map %s", configs.DataPlaneConfigMapAsString()),
-							`configmaps "knative-eventing" not found`,
+						reconcilertesting.WithTriggerBrokerFailed(
+							"Broker not found in data plane map",
+							fmt.Sprintf("config map: %s", configs.DataPlaneConfigMapAsString()),
 						),
 					),
 				},