Repository: knative-sandbox/eventing-kafka-broker
Pull request #1558 (2021-12-01 09:23:28)

Channel to use new configmap format for bootstrap servers
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Channel to use new configmap format for bootstrap servers
- Nothing else is used from configmap at this moment (no auth etc.)

<!--
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-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1558#" 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":[]} -->
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/channel/channel.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1558/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1465974989776752640/artifacts/line-cov.html#file0) | 77.3% | 77.2% | -0.1
[control-plane/pkg/reconciler/kafka/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1558/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1465974989776752640/artifacts/line-cov.html#file1) | 94.9% | 84.8% | -10.1

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1558](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (06d9152) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/1a38db023c18ee55a6bdad88765ca6a601b93431?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1a38db0) will **increase** coverage by `6.74%`.
> The diff coverage is `n/a`.

> :exclamation: Current head 06d9152 differs from pull request most recent head f630a8d. Consider uploading reports for the commit f630a8d to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/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/1558?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    #1558      +/-   ##
============================================
+ Coverage     75.21%   81.96%   +6.74%     
+ Complexity      592      590       -2     
============================================
  Files           107       66      -41     
  Lines          4124     1924    -2200     
  Branches        170      170              
============================================
- Hits           3102     1577    -1525     
+ Misses          764      258     -506     
+ Partials        258       89     -169     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.96% <ø> (-0.11%)` | :arrow_down: |

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/1558?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.50% <0.00%> (-6.25%)` | :arrow_down: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | | |
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | |
| [control-plane/pkg/reconciler/sink/auth\_provider.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2F1dGhfcHJvdmlkZXIuZ28=) | | |
| [control-plane/pkg/core/config/resource.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvcmVzb3VyY2UuZ28=) | | |
| [control-plane/pkg/security/config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvY29uZmlnLmdv) | | |
| [control-plane/pkg/core/config/egress.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvZWdyZXNzLmdv) | | |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| ... and [32 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558/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/1558?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/1558?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1a38db0...f630a8d](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1558?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: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1558#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1558#pullrequestreview-820001804" title="LGTM">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)~~ [aliok,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/config/100-channel/100-kafka-channel-configmap.yaml b/control-plane/config/100-channel/100-kafka-channel-configmap.yaml
new file mode 100644
index 000000000..3f9bab504
--- /dev/null
+++ b/control-plane/config/100-channel/100-kafka-channel-configmap.yaml
@@ -0,0 +1,25 @@
+---
+
+# 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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kafka-channel-config
+  namespace: knative-eventing
+  labels:
+    kafka.eventing.knative.dev/release: devel
+data:
+  bootstrap.servers: "my-cluster-kafka-bootstrap.kafka:9092"
diff --git a/control-plane/config/200-controller/500-controller.yaml b/control-plane/config/200-controller/500-controller.yaml
index 41270a6ce..bc327ce36 100644
--- a/control-plane/config/200-controller/500-controller.yaml
+++ b/control-plane/config/200-controller/500-controller.yaml
@@ -92,7 +92,7 @@ spec:
             - name: BROKER_GENERAL_CONFIG_MAP_NAME
               value: kafka-broker-config
             - name: CHANNEL_GENERAL_CONFIG_MAP_NAME
-              value: kafka-broker-config
+              value: kafka-channel-config
             - name: SINK_GENERAL_CONFIG_MAP_NAME
               value: kafka-broker-config
             - name: SOURCE_GENERAL_CONFIG_MAP_NAME
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index ff39bfd90..6422d04aa 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 	"strconv"
-	"strings"
 
 	"github.com/Shopify/sarama"
 
@@ -40,9 +39,7 @@ import (
 	v1 "knative.dev/eventing/pkg/apis/duck/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
-	commonconfig "knative.dev/eventing-kafka/pkg/common/config"
 	"knative.dev/eventing-kafka/pkg/common/constants"
-	commonsarama "knative.dev/eventing-kafka/pkg/common/kafka/sarama"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -116,19 +113,15 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	}
 	logger.Debug("configmap read", zap.Any("configmap", channelConfigMap))
 
-	// parse the config
-	eventingKafkaSettings, err := commonsarama.LoadEventingKafkaSettings(channelConfigMap.Data)
-	if err != nil {
-		return statusConditionManager.FailedToResolveConfig(err)
-	}
-	logger.Debug("config parsed", zap.Any("eventingKafkaSettings", eventingKafkaSettings))
-
 	if err := r.TrackConfigMap(channelConfigMap, channel); err != nil {
 		return fmt.Errorf("failed to track broker config: %w", err)
 	}
 
 	// get topic config
-	topicConfig := r.topicConfig(logger, eventingKafkaSettings, channel)
+	topicConfig, err := r.topicConfig(logger, channelConfigMap, channel)
+	if err != nil {
+		return statusConditionManager.FailedToResolveConfig(err)
+	}
 	logger.Debug("topic config resolved", zap.Any("config", topicConfig))
 	statusConditionManager.ConfigResolved()
 
@@ -357,15 +350,12 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	}
 	logger.Debug("configmap read", zap.Any("configmap", channelConfigMap))
 
-	// parse the config
-	eventingKafkaSettings, err := commonsarama.LoadEventingKafkaSettings(channelConfigMap.Data)
+	// get topic config
+	topicConfig, err := r.topicConfig(logger, channelConfigMap, channel)
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to resolve channel config: %w", err)
 	}
-	logger.Debug("config parsed", zap.Any("eventingKafkaSettings", eventingKafkaSettings))
 
-	// get topic config
-	topicConfig := r.topicConfig(logger, eventingKafkaSettings, channel)
 	logger.Debug("topic config resolved", zap.Any("config", topicConfig))
 
 	// get the secret to access Kafka
@@ -500,15 +490,20 @@ func (r *Reconciler) channelConfigMap() (*corev1.ConfigMap, error) {
 	// TODO: do we want to support namespaced channels? they're not supported at the moment.
 
 	namespace := system.Namespace()
-	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(constants.SettingsConfigMapName)
+	cm, err := r.ConfigMapLister.ConfigMaps(namespace).Get(r.Env.GeneralConfigMapName)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, constants.SettingsConfigMapName, err)
+		return nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, r.Env.GeneralConfigMapName, err)
 	}
 
 	return cm, nil
 }
 
-func (r *Reconciler) topicConfig(logger *zap.Logger, eventingKafkaConfig *commonconfig.EventingKafkaConfig, channel *messagingv1beta1.KafkaChannel) *kafka.TopicConfig {
+func (r *Reconciler) topicConfig(logger *zap.Logger, cm *corev1.ConfigMap, channel *messagingv1beta1.KafkaChannel) (*kafka.TopicConfig, error) {
+	bootstrapServers, err := kafka.BootstrapServersFromConfigMap(logger, cm)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get bootstrapServers from configmap: %w - ConfigMap data: %v", err, cm.Data)
+	}
+
 	// Parse & Format the RetentionDuration into Sarama retention.ms string
 	retentionDuration, err := channel.Spec.ParseRetentionDuration()
 	if err != nil {
@@ -526,8 +521,8 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, eventingKafkaConfig *common
 				constants.KafkaTopicConfigRetentionMs: &retentionMillisString,
 			},
 		},
-		BootstrapServers: strings.Split(eventingKafkaConfig.Kafka.Brokers, ","),
-	}
+		BootstrapServers: bootstrapServers,
+	}, nil
 }
 
 func (r *Reconciler) secret(ctx context.Context, channelConfig *corev1.ConfigMap) (*corev1.Secret, error) {
diff --git a/control-plane/pkg/reconciler/kafka/topic.go b/control-plane/pkg/reconciler/kafka/topic.go
index 6852a4063..e15a5ce9d 100644
--- a/control-plane/pkg/reconciler/kafka/topic.go
+++ b/control-plane/pkg/reconciler/kafka/topic.go
@@ -52,12 +52,32 @@ func TopicConfigFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) (*TopicC
 	logger.Debug("topic config from configmap",
 		zap.Int32("numPartitions", config.TopicDetail.NumPartitions),
 		zap.Int16("replicationFactor", config.TopicDetail.ReplicationFactor),
-		zap.Any("replicationFactor", config.BootstrapServers),
+		zap.Any("bootstrapServers", config.BootstrapServers),
 	)
 
 	return config, nil
 }
 
+func BootstrapServersFromConfigMap(logger *zap.Logger, cm *corev1.ConfigMap) ([]string, error) {
+	topicConfig, err := buildTopicConfigFromConfigMap(cm)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse config map %s/%s: %w", cm.Namespace, cm.Name, err)
+	}
+
+	if len(topicConfig.BootstrapServers) == 0 {
+		return nil, fmt.Errorf(
+			"invalid configuration bootstrapServers: %s",
+			topicConfig.BootstrapServers,
+		)
+	}
+
+	logger.Debug("bootstrap servers from configmap",
+		zap.Any("bootstrapServers", topicConfig.BootstrapServers),
+	)
+
+	return topicConfig.BootstrapServers, nil
+}
+
 func buildTopicConfigFromConfigMap(cm *corev1.ConfigMap) (*TopicConfig, error) {
 	topicDetail := sarama.TopicDetail{}
 


Test patch:
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index a19edb07e..58ab37815 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/pkg/network"
 
 	corev1 "k8s.io/api/core/v1"
@@ -49,37 +50,8 @@ import (
 	messagingv1beta "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
 	messagingv1beta1kafkachannelreconciler "knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel"
-	"knative.dev/eventing-kafka/pkg/common/constants"
 )
 
-// TODO: are we using all of these?
-var configKafka = map[string]string{
-	"version": "1.0.0",
-	"sarama": `
-enableLogging: false
-config: |
-  Version: 2.0.0 # Kafka Version Compatibility From Sarama's Supported List (Major.Minor.Patch)
-  Admin:
-    Timeout: 10000000000  # 10 seconds
-  Net:
-    KeepAlive: 30000000000  # 30 seconds
-    MaxOpenRequests: 1 # Set to 1 for use with Idempotent Producer
-    TLS:
-      Enable: true
-    SASL:
-      Enable: true
-      Version: 1
-  Metadata:
-    RefreshFrequency: 300000000000  # 5 minutes
-`,
-	"eventing-kafka": `
-kafka:
-  authSecretName: kafka-cluster
-  authSecretNamespace: knative-eventing
-  brokers: kafka:9092
-`,
-}
-
 const (
 	testProber = "testProber"
 
@@ -97,7 +69,7 @@ var finalizerUpdatedEvent = Eventf(
 var DefaultEnv = &config.Env{
 	DataPlaneConfigMapNamespace: "knative-eventing",
 	DataPlaneConfigMapName:      "kafka-channel-channels-subscriptions",
-	GeneralConfigMapName:        "kafka-broker-config",
+	GeneralConfigMapName:        "kafka-channel-config",
 	IngressName:                 "kafka-channel-ingress",
 	SystemNamespace:             "knative-eventing",
 	DataPlaneConfigFormat:       base.Json,
@@ -144,7 +116,9 @@ func TestReconcileKind(t *testing.T) {
 				NewChannel(
 					WithInitKafkaChannelConditions,
 					WithDeletedTimeStamp),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 		},
@@ -161,7 +135,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -227,7 +203,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -295,7 +273,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -363,7 +343,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
@@ -434,7 +416,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
@@ -505,7 +489,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
@@ -579,7 +565,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
@@ -651,7 +639,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key:                     testKey,
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
@@ -685,7 +675,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key:                     testKey,
 			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
@@ -711,7 +703,7 @@ func TestReconcileKind(t *testing.T) {
 			},
 		},
 		{
-			Name: "config-kafka not resolved",
+			Name: "channel configmap not resolved",
 			Objects: []runtime.Object{
 				NewChannel(),
 				NewService(),
@@ -732,7 +724,48 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewChannel(
 						WithInitKafkaChannelConditions,
 						StatusDataPlaneAvailable,
-						StatusConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), constants.SettingsConfigMapName, constants.SettingsConfigMapName)),
+						StatusConfigNotParsed(fmt.Sprintf(`failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName)),
+					),
+				},
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), DefaultEnv.GeneralConfigMapName, DefaultEnv.GeneralConfigMapName),
+				),
+			},
+		},
+		{
+			Name: "channel configmap does not have bootstrap servers",
+			Objects: []runtime.Object{
+				NewChannel(),
+				NewService(),
+				ChannelReceiverPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				ChannelDispatcherPod(env.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					"foo": "bar",
+				}),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the channel namespace with configmap namespace, so skip it
+			WantErr:                 true,
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewChannel(
+						WithInitKafkaChannelConditions,
+						StatusDataPlaneAvailable,
+						StatusConfigNotParsed("unable to get bootstrapServers from configmap: invalid configuration bootstrapServers: [] - ConfigMap data: map[foo:bar]"),
 					),
 				},
 			},
@@ -744,12 +777,12 @@ func TestReconcileKind(t *testing.T) {
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
-					fmt.Sprintf(`failed to get contract configuration: failed to get configmap %s/%s: configmap %q not found`, system.Namespace(), constants.SettingsConfigMapName, constants.SettingsConfigMapName),
+					"failed to get contract configuration: unable to get bootstrapServers from configmap: invalid configuration bootstrapServers: [] - ConfigMap data: map[foo:bar]",
 				),
 			},
 		},
 		{
-			Name: "config-kafka not readable",
+			Name: "channel configmap has blank bootstrap servers",
 			Objects: []runtime.Object{
 				NewChannel(),
 				NewService(),
@@ -761,9 +794,8 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, map[string]string{
-					"version":        "1.0.0",
-					"eventing-kafka": `123123`,
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: "",
 				}),
 			},
 			Key:                     testKey,
@@ -774,7 +806,7 @@ func TestReconcileKind(t *testing.T) {
 					Object: NewChannel(
 						WithInitKafkaChannelConditions,
 						StatusDataPlaneAvailable,
-						StatusConfigNotParsed("ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: error unmarshaling JSON: while decoding JSON: json: cannot unmarshal number into Go value of type config.EventingKafkaConfig : 123123"),
+						StatusConfigNotParsed("unable to get bootstrapServers from configmap: invalid configuration bootstrapServers: [] - ConfigMap data: map[bootstrap.servers:]"),
 					),
 				},
 			},
@@ -786,7 +818,7 @@ func TestReconcileKind(t *testing.T) {
 				Eventf(
 					corev1.EventTypeWarning,
 					"InternalError",
-					"failed to get contract configuration: ConfigMap's eventing-kafka value could not be converted to an EventingKafkaConfig struct: error unmarshaling JSON: while decoding JSON: json: cannot unmarshal number into Go value of type config.EventingKafkaConfig : 123123",
+					"failed to get contract configuration: unable to get bootstrapServers from configmap: invalid configuration bootstrapServers: [] - ConfigMap data: map[bootstrap.servers:]",
 				),
 			},
 		},
@@ -807,7 +839,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			OtherTestData: map[string]interface{}{
 				TestExpectedDataNumPartitions: int32(3),
@@ -900,7 +934,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -966,7 +1002,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, nil),
 			},
 			Key: testKey,
@@ -1030,7 +1068,9 @@ func TestReconcileKind(t *testing.T) {
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				NewConfigMapWithTextData(system.Namespace(), constants.SettingsConfigMapName, configKafka),
+				NewConfigMapWithTextData(system.Namespace(), DefaultEnv.GeneralConfigMapName, map[string]string{
+					kafka.BootstrapServersConfigMapKey: ChannelBootstrapServers,
+				}),
 				NewConfigMapWithBinaryData(&env, []byte("corrupt")),
 			},
 			Key:                     testKey,
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index c6de5977c..675b518da 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -40,7 +40,7 @@ const (
 	ChannelName             = "kc"
 	ChannelNamespace        = "test-nc"
 	ChannelUUID             = "c1234567-8901-2345-6789-123456789101"
-	ChannelBootstrapServers = "kafka:9092"
+	ChannelBootstrapServers = "kafka-1:9092,kafka-2:9093"
 
 	Subscription1Name     = "sub-1"
 	Subscription2Name     = "sub-2"