Make configmap reading reusable (#1551)

* Rename `configFromConfigMap` to `TopicConfigFromConfigMap`

* Characterization test for broker_config

* Make topicConfig reading code reusable, do the broker specific topicConfig validation in broker package

* Make topicConfig reading code reusable, do the broker specific topicConfig validation in broker package

* Validate topicConfig from configmap in-place

* Fix IDE method extraction problems

* Fix issues after rebase

control-plane/pkg/reconciler/broker/broker.go

@@ -364,12 +364,12 @@ func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*
 		return nil, nil, fmt.Errorf("failed to get configmap %s/%s: %w", namespace, broker.Spec.Config.Name, err)
 	}
 
-	brokerConfig, err := configFromConfigMap(logger, cm)
+	topicConfig, err := kafka.TopicConfigFromConfigMap(logger, cm)
 	if err != nil {
-		return nil, cm, err
+		return nil, cm, fmt.Errorf("unable to build topic config from configmap: %w - ConfigMap data: %v", err, cm.Data)
 	}
 
-	return brokerConfig, cm, nil
+	return topicConfig, cm, nil
 }
 
 func (r *Reconciler) defaultTopicDetail() sarama.TopicDetail {
@@ -431,8 +431,13 @@ func (r *Reconciler) ConfigMapUpdated(ctx context.Context) func(configMap *corev
 
 	return func(configMap *corev1.ConfigMap) {
 
-		topicConfig, err := configFromConfigMap(logger, 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
 		}
 
@@ -472,7 +477,7 @@ func (r *Reconciler) getDefaultBootstrapServersOrFail() ([]string, error) {
 	defer r.bootstrapServersLock.RUnlock()
 
 	if len(r.bootstrapServers) == 0 {
-		return nil, fmt.Errorf("no %s provided", BootstrapServersConfigMapKey)
+		return nil, fmt.Errorf("no %s provided", kafka.BootstrapServersConfigMapKey)
 	}
 
 	return r.bootstrapServers, nil