Repository: banzaicloud/kafka-operator
Pull request #542 (2021-01-28 07:37:03)

refactor: use properties lib for handlig configs
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | yes
| API breaks?     | no
| Deprecations?   | no
| License         | Apache 2.0

### What's in this PR?
Improve the way of handling Java Properties configuration by using `github.com/banzaicloud/kafka-operator/properties` library.

### Why?
The `github.com/banzaicloud/kafka-operator/properties` is a more sophisticated solution for handling Java Properties.

- [x] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/developer-guide/blob/master/docs/coding-style/error-handling-guide.md)
- [x] Logging code meets the guideline
- [x] User guide and development docs updated (if needed)

### To Do
- [x] Update PR after #540 got merged
- [x] Update PR after #543 got merged
- [x] Implement `Sort` and `Diff` methods/functions in `github.com/banzaicloud/kafka-operator/properties`


chrisgacsal: > Thanks for the work @chrisgacsal, the properties lib is very powerful and increases the quality of the codebase by a lot!
> 
> A generic comment:
> I see a lot of `config.Set(key, value, "")`. Though it's not a big effort, to put that `, ""` as a third parameter every single time, I think it may worth to have these functions as setting
> the config without comment is much frequently used than the other:
> 
> ```go
> func Set(key string, value interface{})
> func SetWithComment(key string, value interface{}, comment string)
> ```

Agreed. I've been thinking about this as well. Will add `SetWithComment` to the properties lib.
chrisgacsal: @hi-im-aren @adamantal @stoader @baluchicken Thank you guys for reviewing the PR. I have incorporated your suggestions. Please take another look on it. Many thanks!

Patch:
diff --git a/go.mod b/go.mod
index 3483e2b6..0bee90ca 100644
--- a/go.mod
+++ b/go.mod
@@ -3,13 +3,14 @@ module github.com/banzaicloud/kafka-operator
 go 1.15
 
 require (
-	emperror.dev/errors v0.7.0
+	emperror.dev/errors v0.8.0
 	github.com/Shopify/sarama v1.27.1
 	github.com/banzaicloud/bank-vaults/pkg/sdk v0.3.1
 	github.com/banzaicloud/istio-client-go v0.0.9
 	github.com/banzaicloud/istio-operator v0.0.0-20200407070503-3f7dc6953a7b
 	github.com/banzaicloud/k8s-objectmatcher v1.4.1
 	github.com/banzaicloud/kafka-operator/api v0.0.0
+	github.com/banzaicloud/kafka-operator/properties v0.0.0
 	github.com/envoyproxy/go-control-plane v0.9.7
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
 	github.com/go-logr/logr v0.1.0
@@ -33,12 +34,10 @@ require (
 	github.com/onsi/gomega v1.10.1
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
 	github.com/pavel-v-chernykh/keystore-go v2.1.0+incompatible
-	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.4.1 // indirect
 	github.com/prometheus/common v0.9.1
 	github.com/spf13/cast v1.3.1 // indirect
 	github.com/spf13/viper v1.7.1 // indirect
-	go.uber.org/atomic v1.5.1 // indirect
 	go.uber.org/zap v1.10.0
 	google.golang.org/appengine v1.6.6 // indirect
 	google.golang.org/grpc v1.27.1 // indirect
@@ -50,4 +49,7 @@ require (
 	sigs.k8s.io/controller-runtime v0.6.3
 )
 
-replace github.com/banzaicloud/kafka-operator/api => ./pkg/sdk
+replace (
+	github.com/banzaicloud/kafka-operator/api => ./pkg/sdk
+	github.com/banzaicloud/kafka-operator/properties => ./properties
+)
diff --git a/go.sum b/go.sum
index 28cc75d6..09b3a98d 100644
--- a/go.sum
+++ b/go.sum
@@ -19,6 +19,8 @@ code.cloudfoundry.org/gofileutils v0.0.0-20170111115228-4d0c80011a0f/go.mod h1:s
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 emperror.dev/errors v0.7.0 h1:vf0gZ0j9BJCXQYm0M21DPKeXy9PUjvX1YHXhNzIsqOY=
 emperror.dev/errors v0.7.0/go.mod h1:X4dljzQehaz3WfBKc6c7bR+ve2ZsRzbBkFBF+HTcW0M=
+emperror.dev/errors v0.8.0 h1:4lycVEx0sdJkwDUfQ9pdu6SR0x7rgympt5f4+ok8jDk=
+emperror.dev/errors v0.8.0/go.mod h1:YcRvLPh626Ubn2xqtoprejnA5nFha+TJ+2vew48kWuE=
 git.apache.org/thrift.git v0.12.0/go.mod h1:fPE2ZNJGynbRyZ4dJvy6G277gSllfV2HJqblrnkyeyg=
 github.com/Azure/azure-sdk-for-go v32.5.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go v36.2.0+incompatible h1:09cv2WoH0g6jl6m2iT+R9qcIPZKhXEL0sbmLhxP895s=
@@ -1029,11 +1031,13 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.5.0 h1:OI5t8sDa1Or+q8AeE+yKeB/SDYioSHAgcVljj9JIETY=
 go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
-go.uber.org/atomic v1.5.1 h1:rsqfU5vBkVknbhUGbAUwQKR2H4ItV8tjJ+6kJX4cxHM=
-go.uber.org/atomic v1.5.1/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ=
+go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
+go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
 go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
 go.uber.org/multierr v1.4.0 h1:f3WCSC2KzAcBXGATIxAB1E2XuCpNU255wNKZ505qi3E=
 go.uber.org/multierr v1.4.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
+go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee h1:0mgffUl7nfd+FpvXMVz4IDEaUSmT1ysygQC7qYo7sG4=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM=
diff --git a/pkg/k8sutil/resource.go b/pkg/k8sutil/resource.go
index 5bde247d..ef8f49e6 100644
--- a/pkg/k8sutil/resource.go
+++ b/pkg/k8sutil/resource.go
@@ -34,8 +34,8 @@ import (
 
 	certv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1alpha2"
 
-	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
 // Reconcile reconciles K8S resources
@@ -133,8 +133,20 @@ func Reconcile(log logr.Logger, client runtimeClient.Client, desired runtime.Obj
 			case *corev1.ConfigMap:
 				// Only update status when configmap belongs to broker
 				if id, ok := desired.(*corev1.ConfigMap).Labels["brokerId"]; ok {
-					currentConfigs := util.ParsePropertiesFormat(current.(*corev1.ConfigMap).Data[kafka.ConfigPropertyName])
-					desiredConfigs := util.ParsePropertiesFormat(desired.(*corev1.ConfigMap).Data[kafka.ConfigPropertyName])
+					currentConfigs, err := properties.NewFromString(current.(*corev1.ConfigMap).Data[kafka.ConfigPropertyName])
+					if err != nil {
+						return errors.WrapWithDetails(err, "could not parse the current configuration for broker", "brokerId", id)
+					}
+
+					desiredConfigs, err := properties.NewFromString(desired.(*corev1.ConfigMap).Data[kafka.ConfigPropertyName])
+					if err != nil {
+						return errors.WrapWithDetails(err, "could not parse the current configuration for broker", "brokerId", id)
+					}
+
+					// Check if there is drift in the configuration and return in case there is none
+					if currentConfigs.Equal(desiredConfigs) {
+						return nil
+					}
 
 					var statusErr error
 					// if only per broker configs are changed, do not trigger rolling upgrade by setting ConfigOutOfSync status
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index af6f45d1..a97be708 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -28,11 +28,44 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	kafkautils "github.com/banzaicloud/kafka-operator/pkg/util/kafka"
 	zookeeperutils "github.com/banzaicloud/kafka-operator/pkg/util/zookeeper"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 
 	corev1 "k8s.io/api/core/v1"
 )
 
 func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logger) runtime.Object {
+	ccConfig := properties.NewProperties()
+
+	// Add base Cruise Control configuration
+	conf, err := properties.NewFromString(r.KafkaCluster.Spec.CruiseControlConfig.Config)
+	if err != nil {
+		log.Error(err, "parsing Cruise Control configuration failed", "config", r.KafkaCluster.Spec.CruiseControlConfig.Config)
+	}
+	ccConfig.Merge(conf)
+
+	// Set bootstrap servers
+	bootstrapServers := fmt.Sprintf("%s:%d",
+		generateBootstrapServer(r.KafkaCluster.Spec.HeadlessServiceEnabled, r.KafkaCluster.Name),
+		generateBootstrapServerPort(log, r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
+	)
+	if err = ccConfig.Set("bootstrap.servers", bootstrapServers); err != nil {
+		log.Error(err, "setting bootstrap.servers in Cruise Control configuration failed", "config", bootstrapServers)
+	}
+
+	// Add Zookeeper configuration
+	zkConnect := zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath())
+	if err = ccConfig.Set("zookeeper.connect", zkConnect); err != nil {
+		log.Error(err, "setting zookeeper.connect in Cruise Control configuration failed", "config", zkConnect)
+	}
+
+	// Add SSL configuration
+	sslConf := generateSSLConfig(&r.KafkaCluster.Spec.ListenersConfig, clientPass, log)
+	if sslConf.Len() != 0 {
+		ccConfig.Merge(sslConf)
+	}
+
+	ccConfig.Sort()
+
 	configMap := &corev1.ConfigMap{
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(configAndVolumeNameTemplate, r.KafkaCluster.Name),
@@ -40,34 +73,36 @@ func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logge
 			r.KafkaCluster,
 		),
 		Data: map[string]string{
-			"cruisecontrol.properties": r.KafkaCluster.Spec.CruiseControlConfig.Config + fmt.Sprintf(`
-# The Kafka cluster to control.
-bootstrap.servers=%s:%d
-# The zookeeper connect of the Kafka cluster
-zookeeper.connect=%s
-`, generateBootstrapServer(r.KafkaCluster.Spec.HeadlessServiceEnabled, r.KafkaCluster.Name),
-				generateBootstrapServerPort(log, r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
-				zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath())) +
-				generateSSLConfig(&r.KafkaCluster.Spec.ListenersConfig, clientPass),
-			"capacity.json":       capacityConfig,
-			"clusterConfigs.json": r.KafkaCluster.Spec.CruiseControlConfig.ClusterConfig,
-			"log4j.properties":    r.KafkaCluster.Spec.CruiseControlConfig.GetCCLog4jConfig(),
+			"cruisecontrol.properties": ccConfig.String(),
+			"capacity.json":            capacityConfig,
+			"clusterConfigs.json":      r.KafkaCluster.Spec.CruiseControlConfig.ClusterConfig,
+			"log4j.properties":         r.KafkaCluster.Spec.CruiseControlConfig.GetCCLog4jConfig(),
 		},
 	}
 	return configMap
 }
 
-func generateSSLConfig(l *v1beta1.ListenersConfig, clientPass string) (res string) {
+func generateSSLConfig(l *v1beta1.ListenersConfig, clientPass string, log logr.Logger) *properties.Properties {
+	sslConf := properties.NewProperties()
+
 	if l.SSLSecrets != nil && util.IsSSLEnabledForInternalCommunication(l.InternalListeners) {
-		res = fmt.Sprintf(`
-security.protocol=SSL
-ssl.truststore.location=/var/run/secrets/java.io/keystores/%s
-ssl.keystore.location=/var/run/secrets/java.io/keystores/%s
-ssl.keystore.password=%s
-ssl.truststore.password=%s
-`, v1alpha1.TLSJKSTrustStore, v1alpha1.TLSJKSKeyStore, clientPass, clientPass)
-	}
-	return
+		if err := sslConf.Set("security.protocol", "SSL"); err != nil {
+			log.Error(err, "settings security.protocol in Cruise Control configuration failed")
+		}
+		if err := sslConf.Set("ssl.truststore.location", v1alpha1.TLSJKSTrustStore); err != nil {
+			log.Error(err, "settings ssl.truststore.location in Cruise Control configuration failed")
+		}
+		if err := sslConf.Set("ssl.keystore.location", v1alpha1.TLSJKSKeyStore); err != nil {
+			log.Error(err, "settings ssl.keystore.location in Cruise Control configuration failed")
+		}
+		if err := sslConf.Set("ssl.keystore.password", clientPass); err != nil {
+			log.Error(err, "settings ssl.keystore.password in Cruise Control configuration failed")
+		}
+		if err := sslConf.Set("ssl.truststore.password", clientPass); err != nil {
+			log.Error(err, "settings ssl.truststore.password in Cruise Control configuration failed")
+		}
+	}
+	return sslConf
 }
 
 func generateBootstrapServer(headlessEnabled bool, clusterName string) string {
diff --git a/pkg/resources/cruisecontrol/topicManager.go b/pkg/resources/cruisecontrol/topicManager.go
index ff590c8b..fd6ca777 100644
--- a/pkg/resources/cruisecontrol/topicManager.go
+++ b/pkg/resources/cruisecontrol/topicManager.go
@@ -17,14 +17,14 @@ package cruisecontrol
 import (
 	"context"
 	"fmt"
-	"strconv"
 
+	"emperror.dev/errors"
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
-	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/banzaicloud/kafka-operator/pkg/webhook"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 	"github.com/go-logr/logr"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/types"
@@ -71,12 +71,15 @@ func newCruiseControlTopic(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaTopic {
 }
 
 func generateCCTopic(cluster *v1beta1.KafkaCluster, client client.Client, log logr.Logger) error {
-	readOnlyConfigProperties := util.ParsePropertiesFormat(cluster.Spec.ReadOnlyConfig)
+	readOnlyConfigProperties, err := properties.NewFromString(cluster.Spec.ReadOnlyConfig)
+	if err != nil {
+		return errors.WrapIf(err, "could not parse broker config")
+	}
 
 	// for compatibility reasons the only case when we let CC to create its own kafka topics is
 	// when we enable the creation explicitly
-	if propertyValue, present := readOnlyConfigProperties[ccMetricTopicAutoCreate]; present {
-		if autoCreate, err := strconv.ParseBool(propertyValue); err != nil {
+	if autoCreateProperty, present := readOnlyConfigProperties.Get(ccMetricTopicAutoCreate); present {
+		if autoCreate, err := autoCreateProperty.Bool(); err != nil {
 			return err
 		} else if autoCreate {
 			log.Info("CruiseControl topic has been created by CruiseControl")
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index ab39841e..3c42c1f5 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -15,15 +15,11 @@
 package kafka
 
 import (
-	"bytes"
 	"fmt"
-	"sort"
 	"strings"
-	"text/template"
 
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
-	"github.com/imdario/mergo"
 
 	corev1 "k8s.io/api/core/v1"
 
@@ -33,74 +29,105 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
 	zookeeperutils "github.com/banzaicloud/kafka-operator/pkg/util/zookeeper"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
-var kafkaConfigTemplate = `
-{{ .ListenerConfig }}
-{{ if .ControlPlaneListener }}
-control.plane.listener.name={{ .ControlPlaneListener }}
-{{ end }}
-
-zookeeper.connect={{ .ZookeeperConnectString }}
-
-{{ if .KafkaCluster.Spec.ListenersConfig.SSLSecrets }}
-
-ssl.keystore.location={{ .ServerKeystorePath }}/{{ .KeyStoreFile }}
-ssl.truststore.location={{ .ServerKeystorePath }}/{{ .TrustStoreFile }}
-ssl.keystore.password={{ .ServerKeystorePassword }}
-ssl.truststore.password={{ .ServerKeystorePassword }}
-ssl.client.auth=required
-
-{{ if .SSLEnabledForInternalCommunication }}
-
-cruise.control.metrics.reporter.security.protocol=SSL
-cruise.control.metrics.reporter.ssl.truststore.location={{ .ClientKeystorePath }}/{{ .TrustStoreFile }}
-cruise.control.metrics.reporter.ssl.keystore.location={{ .ClientKeystorePath }}/{{ .KeyStoreFile }}
-cruise.control.metrics.reporter.ssl.keystore.password={{ .ClientKeystorePassword }}
-
-{{ end }}
-{{ end }}
-
-metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
-cruise.control.metrics.reporter.bootstrap.servers={{ .CruiseControlBootstrapServers }}
-cruise.control.metrics.reporter.kubernetes.mode=true
-broker.id={{ .Id }}
-
-{{ if .StorageConfig }}
-log.dirs={{ .StorageConfig }}
-{{ end }}
-
-{{ .AdvertisedListenersConfig }}
-
-{{ if .SuperUsers }}
-super.users={{ .SuperUsers }}
-{{ end }}
-`
-
-func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
-	var out bytes.Buffer
-	t := template.Must(template.New("bConfig-config").Parse(kafkaConfigTemplate))
-	if err := t.Execute(&out, map[string]interface{}{
-		"KafkaCluster":                       r.KafkaCluster,
-		"Id":                                 id,
-		"ListenerConfig":                     generateListenerSpecificConfig(&r.KafkaCluster.Spec.ListenersConfig, log),
-		"SSLEnabledForInternalCommunication": r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil && util.IsSSLEnabledForInternalCommunication(r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
-		"ZookeeperConnectString":             zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath()),
-		"CruiseControlBootstrapServers":      getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
-		"StorageConfig":                      generateStorageConfig(bConfig.StorageConfigs),
-		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses),
-		"SuperUsers":                         strings.Join(generateSuperUsers(superUsers), ";"),
-		"ServerKeystorePath":                 serverKeystorePath,
-		"ClientKeystorePath":                 clientKeystorePath,
-		"KeyStoreFile":                       v1alpha1.TLSJKSKeyStore,
-		"TrustStoreFile":                     v1alpha1.TLSJKSTrustStore,
-		"ServerKeystorePassword":             serverPass,
-		"ClientKeystorePassword":             clientPass,
-		"ControlPlaneListener":               generateControlPlaneListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
-	}); err != nil {
-		log.Error(err, "error occurred during parsing the config template")
+func (r *Reconciler) getConfigProperties(bConfig *v1beta1.BrokerConfig, id int32, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) *properties.Properties {
+	config := properties.NewProperties()
+
+	// Add listener configuration
+	listenerConf := generateListenerSpecificConfig(&r.KafkaCluster.Spec.ListenersConfig, log)
+	config.Merge(listenerConf)
+
+	// Add listener configuration
+	advertisedListenerConf := generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses)
+	if len(advertisedListenerConf) > 0 {
+		if err := config.Set("advertised.listeners", advertisedListenerConf); err != nil {
+			log.Error(err, "setting advertised.listeners in broker configuration resulted an error")
+		}
+	}
+
+	// Add control plane listener
+	cclConf := generateControlPlaneListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners)
+	if cclConf != "" {
+		if err := config.Set("control.plane.listener.name", cclConf); err != nil {
+			log.Error(err, "setting control.plane.listener.name parameter in broker configuration resulted an error")
+		}
+	}
+
+	// Add Zookeeper configuration
+	if err := config.Set("zookeeper.connect", zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath())); err != nil {
+		log.Error(err, "setting zookeeper.connect parameter in broker configuration resulted an error")
+	}
+
+	// Add SSL configuration
+	if r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil {
+		if err := config.Set("ssl.keystore.location", serverKeystorePath+"/"+v1alpha1.TLSJKSKeyStore); err != nil {
+			log.Error(err, "setting ssl.keystore.location parameter in broker configuration resulted an error")
+		}
+		if err := config.Set("ssl.truststore.location", serverKeystorePath+"/"+v1alpha1.TLSJKSTrustStore); err != nil {
+			log.Error(err, "setting ssl.truststore.location parameter in broker configuration resulted an error")
+		}
+		if err := config.Set("ssl.keystore.password", serverPass); err != nil {
+			log.Error(err, "setting ssl.keystore.password parameter in broker configuration resulted an error")
+		}
+		if err := config.Set("ssl.truststore.password", serverPass); err != nil {
+			log.Error(err, "setting ssl.truststore.password parameter in broker configuration resulted an error")
+		}
+		if err := config.Set("ssl.client.auth", "required"); err != nil {
+			log.Error(err, "setting ssl.client.auth parameter in broker configuration resulted an error")
+		}
+
+		// Add Cruise Control SSL configuration
+		if util.IsSSLEnabledForInternalCommunication(r.KafkaCluster.Spec.ListenersConfig.InternalListeners) {
+			if err := config.Set("cruise.control.metrics.reporter.security.protocol", "SSL"); err != nil {
+				log.Error(err, "setting cruise.control.metrics.reporter.security.protocol in broker configuration resulted an error")
+			}
+			if err := config.Set("cruise.control.metrics.reporter.ssl.truststore.location", clientKeystorePath+"/"+v1alpha1.TLSJKSTrustStore); err != nil {
+				log.Error(err, "setting cruise.control.metrics.reporter.ssl.truststore.location in broker configuration resulted an error")
+			}
+			if err := config.Set("cruise.control.metrics.reporter.ssl.keystore.location", clientKeystorePath+"/"+v1alpha1.TLSJKSKeyStore); err != nil {
+				log.Error(err, "setting cruise.control.metrics.reporter.ssl.keystore.location parameter in broker configuration resulted an error")
+			}
+			if err := config.Set("cruise.control.metrics.reporter.ssl.keystore.password", clientPass); err != nil {
+				log.Error(err, "setting cruise.control.metrics.reporter.ssl.keystore.password parameter in broker configuration resulted an error")
+			}
+		}
+	}
+
+	// Add Cruise Control Metrics Reporter configuration
+	if err := config.Set("metric.reporters", "com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"); err != nil {
+		log.Error(err, "setting metric.reporters in broker configuration resulted an error")
 	}
-	return out.String()
+	bootstrapServers := getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled)
+	if err := config.Set("cruise.control.metrics.reporter.bootstrap.servers", bootstrapServers); err != nil {
+		log.Error(err, "setting cruise.control.metrics.reporter.bootstrap.servers in broker configuration resulted an error")
+	}
+	if err := config.Set("cruise.control.metrics.reporter.kubernetes.mode", true); err != nil {
+		log.Error(err, "setting cruise.control.metrics.reporter.kubernetes.mode in broker configuration resulted an error")
+	}
+
+	// Kafka Broker configuration
+	if err := config.Set("broker.id", id); err != nil {
+		log.Error(err, "setting broker.id in broker configuration resulted an error")
+	}
+
+	// Storage configuration
+	storageConf := generateStorageConfig(bConfig.StorageConfigs)
+	if storageConf != "" {
+		if err := config.Set("log.dirs", storageConf); err != nil {
+			log.Error(err, "setting log.dirs in broker configuration resulted an error")
+		}
+	}
+
+	// Add superuser configuration
+	su := strings.Join(generateSuperUsers(superUsers), ";")
+	if su != "" {
+		if err := config.Set("super.users", su); err != nil {
+			log.Error(err, "setting super.users in broker configuration resulted an error")
+		}
+	}
+	return config
 }
 
 func generateSuperUsers(users []string) (suStrings []string) {
@@ -130,14 +157,14 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, ext
 }
 
 func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig,
-	extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList) string {
+	extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList) []string {
 	advertisedListenerConfig := make([]string, 0, len(l.ExternalListeners)+len(l.InternalListeners))
 
 	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, extListenerStatuses)
 	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, intListenerStatuses)
 	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, controllerIntListenerStatuses)
 
-	return fmt.Sprintf("advertised.listeners=%s\n", strings.Join(advertisedListenerConfig, ","))
+	return advertisedListenerConfig
 }
 
 func appendListenerConfigs(advertisedListenerConfig []string, id int32, listenerStatusList map[string]v1beta1.ListenerStatusList) []string {
@@ -173,7 +200,7 @@ func generateControlPlaneListener(iListeners []v1beta1.InternalListenerConfig) s
 	return controlPlaneListener
 }
 
-func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger) string {
+func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger) *properties.Properties {
 
 	var interBrokerListenerName string
 	var securityProtocolMapConfig []string
@@ -198,9 +225,19 @@ func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger)
 		securityProtocolMapConfig = append(securityProtocolMapConfig, fmt.Sprintf("%s:%s", UpperedListenerName, UpperedListenerType))
 		listenerConfig = append(listenerConfig, fmt.Sprintf("%s://:%d", UpperedListenerName, eListener.ContainerPort))
 	}
-	return "listener.security.protocol.map=" + strings.Join(securityProtocolMapConfig, ",") + "\n" +
-		"inter.broker.listener.name=" + interBrokerListenerName + "\n" +
-		"listeners=" + strings.Join(listenerConfig, ",") + "\n"
+
+	config := properties.NewProperties()
+	if err := config.Set("listener.security.protocol.map", securityProtocolMapConfig); err != nil {
+		log.Error(err, "setting listener.security.protocol.map parameter in broker configuration resulted an error")
+	}
+	if err := config.Set("inter.broker.listener.name", interBrokerListenerName); err != nil {
+		log.Error(err, "setting inter.broker.listener.name parameter in broker configuration resulted an error")
+	}
+	if err := config.Set("listeners", listenerConfig); err != nil {
+		log.Error(err, "setting listeners parameter in broker configuration resulted an error")
+	}
+
+	return config
 }
 
 func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32, domain, namespace, crName string, headlessServiceEnabled bool) string {
@@ -221,39 +258,39 @@ func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32,
 }
 
 func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
-	parsedReadOnlyClusterConfig := util.ParsePropertiesFormat(r.KafkaCluster.Spec.ReadOnlyConfig)
-	var parsedReadOnlyBrokerConfig = map[string]string{}
+	// Parse cluster-wide readonly configuration
+	finalBrokerConfig, err := properties.NewFromString(r.KafkaCluster.Spec.ReadOnlyConfig)
+	if err != nil {
+		log.Error(err, "failed to parse readonly cluster configuration")
+	}
 
+	// Parse readonly broker configuration
+	var parsedReadOnlyBrokerConfig *properties.Properties
+	// Find configuration for broker with id
 	for _, broker := range r.KafkaCluster.Spec.Brokers {
 		if broker.Id == id {
-			parsedReadOnlyBrokerConfig = util.ParsePropertiesFormat(broker.ReadOnlyConfig)
+			parsedReadOnlyBrokerConfig, err = properties.NewFromString(broker.ReadOnlyConfig)
+			if err != nil {
+				log.Error(err, fmt.Sprintf("failed to parse readonly broker configuration for broker with id: %d", id))
+			}
 			break
 		}
 	}
 
-	if err := mergo.Merge(&parsedReadOnlyBrokerConfig, parsedReadOnlyClusterConfig); err != nil {
-		log.Error(err, "error occurred during merging readonly configs")
-	}
-
-	//Generate the Complete Configuration for the Broker
-	completeConfigMap := map[string]string{}
-
-	if err := mergo.Merge(&completeConfigMap, util.ParsePropertiesFormat(r.getConfigString(brokerConfig, id, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log))); err != nil {
-		log.Error(err, "error occurred during merging operator generated configs")
-	}
-
-	if err := mergo.Merge(&completeConfigMap, parsedReadOnlyBrokerConfig); err != nil {
-		log.Error(err, "error occurred during merging readOnly config to complete configs")
+	// Merge cluster-wide configuration into broker-level configuration
+	if parsedReadOnlyBrokerConfig != nil {
+		finalBrokerConfig.Merge(parsedReadOnlyBrokerConfig)
 	}
 
-	completeConfig := make([]string, 0, len(completeConfigMap))
+	// Get operator generated configuration
+	opGenConf := r.getConfigProperties(brokerConfig, id, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)
 
-	for key, value := range completeConfigMap {
-		completeConfig = append(completeConfig, fmt.Sprintf("%s=%s", key, value))
+	// Merge operator generated configuration to the final one
+	if opGenConf != nil {
+		finalBrokerConfig.Merge(opGenConf)
 	}
 
-	// We need to sort the config every time to avoid diffs occurred because of ranging through map
-	sort.Strings(completeConfig)
+	finalBrokerConfig.Sort()
 
-	return strings.Join(completeConfig, "\n")
+	return finalBrokerConfig.String()
 }
diff --git a/pkg/resources/kafka/configs.go b/pkg/resources/kafka/configs.go
index ab22a69a..8bcddec1 100644
--- a/pkg/resources/kafka/configs.go
+++ b/pkg/resources/kafka/configs.go
@@ -28,6 +28,8 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
+
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
 func (r *Reconciler) reconcilePerBrokerDynamicConfig(brokerId int32, brokerConfig *v1beta1.BrokerConfig, configMap *corev1.ConfigMap, log logr.Logger) error {
@@ -41,26 +43,29 @@ func (r *Reconciler) reconcilePerBrokerDynamicConfig(brokerId int32, brokerConfi
 		}
 	}()
 
-	fullPerBrokerConfig := util.ParsePropertiesFormat(brokerConfig.Config)
+	fullPerBrokerConfig, err := properties.NewFromString(brokerConfig.Config)
+	if err != nil {
+		return errors.WrapIf(err, "could not parse broker configuration")
+	}
 
 	currentPerBrokerConfigState := r.KafkaCluster.Status.BrokersState[strconv.Itoa(int(brokerId))].PerBrokerConfigurationState
-	if len(fullPerBrokerConfig) == 0 && currentPerBrokerConfigState != v1beta1.PerBrokerConfigOutOfSync {
+	if fullPerBrokerConfig.Len() == 0 && currentPerBrokerConfigState != v1beta1.PerBrokerConfigOutOfSync {
 		return nil
 	}
 
 	// overwrite configs from configmap
-	configsFromConfigMap := util.ParsePropertiesFormat(configMap.Data[kafka.ConfigPropertyName])
+	configsFromConfigMap, err := properties.NewFromString(configMap.Data[kafka.ConfigPropertyName])
+	if err != nil {
+		return errors.WrapIf(err, "could not parse broker configuration from configmap")
+	}
 	for _, perBrokerConfig := range kafka.PerBrokerConfigs {
-		if configValue, ok := configsFromConfigMap[perBrokerConfig]; ok {
-			fullPerBrokerConfig[perBrokerConfig] = configValue
+		if configProperty, ok := configsFromConfigMap.Get(perBrokerConfig); ok {
+			fullPerBrokerConfig.Put(configProperty)
 		}
 	}
 
-	// query the current configs
-	brokerConfigKeys := make([]string, 0, len(fullPerBrokerConfig))
-	for key := range fullPerBrokerConfig {
-		brokerConfigKeys = append(brokerConfigKeys, key)
-	}
+	// query the current config
+	brokerConfigKeys := fullPerBrokerConfig.Keys()
 	response, err := kClient.DescribePerBrokerConfig(brokerId, brokerConfigKeys)
 	if err != nil {
 		return errors.WrapIfWithDetails(err, "could not describe broker config", "brokerId", brokerId)
@@ -76,7 +81,7 @@ func (r *Reconciler) reconcilePerBrokerDynamicConfig(brokerId int32, brokerConfi
 		}
 
 		// validate the config
-		err := kClient.AlterPerBrokerConfig(brokerId, util.ConvertMapStringToMapStringPointer(fullPerBrokerConfig), true)
+		err := kClient.AlterPerBrokerConfig(brokerId, util.ConvertPropertiesToMapStringPointer(fullPerBrokerConfig), true)
 		if err != nil {
 			statusErr := k8sutil.UpdateBrokerStatus(r.Client, []string{strconv.Itoa(int(brokerId))}, r.KafkaCluster, v1beta1.PerBrokerConfigError, log)
 			if statusErr != nil {
@@ -86,7 +91,7 @@ func (r *Reconciler) reconcilePerBrokerDynamicConfig(brokerId int32, brokerConfi
 		}
 
 		// alter the config
-		err = kClient.AlterPerBrokerConfig(brokerId, util.ConvertMapStringToMapStringPointer(fullPerBrokerConfig), false)
+		err = kClient.AlterPerBrokerConfig(brokerId, util.ConvertPropertiesToMapStringPointer(fullPerBrokerConfig), false)
 		if err != nil {
 			return errors.WrapIfWithDetails(err, "could not alter broker config", "brokerId", brokerId)
 		}
@@ -128,34 +133,28 @@ func (r *Reconciler) reconcileClusterWideDynamicConfig(log logr.Logger) error {
 		}
 	}()
 
-	configIdentical := true
-
 	currentConfig, err := kClient.DescribeClusterWideConfig()
 	if err != nil {
 		return errors.WrapIf(err, "could not describe cluster wide broker config")
 	}
-	parsedClusterWideConfig := util.ParsePropertiesFormat(r.KafkaCluster.Spec.ClusterWideConfig)
-	if len(currentConfig) != len(parsedClusterWideConfig) {
-		configIdentical = false
+
+	currentClusterWideConfig, err := util.ConvertConfigEntryListToProperties(currentConfig)
+	if err != nil {
+		return errors.WrapIf(err, "could not convert current cluster-wide config to properties")
 	}
-	if configIdentical {
-		for _, conf := range currentConfig {
-			if val, ok := parsedClusterWideConfig[conf.Name]; ok {
-				if val != conf.Value {
-					configIdentical = false
-					break
-				}
-			}
-		}
+
+	parsedClusterWideConfig, err := properties.NewFromString(r.KafkaCluster.Spec.ClusterWideConfig)
+	if err != nil {
+		return errors.WrapIf(err, "could not parse cluster-wide broker config")
 	}
 
-	if !configIdentical {
-		err = kClient.AlterClusterWideConfig(util.ConvertMapStringToMapStringPointer(parsedClusterWideConfig), true)
+	if !currentClusterWideConfig.Equal(parsedClusterWideConfig) {
+		err = kClient.AlterClusterWideConfig(util.ConvertPropertiesToMapStringPointer(parsedClusterWideConfig), true)
 		if err != nil {
 			return errors.WrapIf(err, "validation of cluster wide config update failed")
 		}
 
-		err = kClient.AlterClusterWideConfig(util.ConvertMapStringToMapStringPointer(parsedClusterWideConfig), false)
+		err = kClient.AlterClusterWideConfig(util.ConvertPropertiesToMapStringPointer(parsedClusterWideConfig), false)
 		if err != nil {
 			return errors.WrapIf(err, "could not alter cluster wide broker config")
 		}
@@ -164,10 +163,14 @@ func (r *Reconciler) reconcileClusterWideDynamicConfig(log logr.Logger) error {
 	return nil
 }
 
-func shouldUpdatePerBrokerConfig(response []*sarama.ConfigEntry, parsedBrokerConfig map[string]string) bool {
+func shouldUpdatePerBrokerConfig(response []*sarama.ConfigEntry, brokerConfig *properties.Properties) bool {
+	if brokerConfig == nil {
+		return false
+	}
+
 	for _, conf := range response {
-		if val, ok := parsedBrokerConfig[conf.Name]; ok {
-			if val != conf.Value {
+		if val, ok := brokerConfig.Get(conf.Name); ok {
+			if val.Value() != conf.Value {
 				return true
 			}
 		}
diff --git a/pkg/util/kafka/common.go b/pkg/util/kafka/common.go
index 4d49d820..fce2a05b 100644
--- a/pkg/util/kafka/common.go
+++ b/pkg/util/kafka/common.go
@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"strings"
 
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 	"github.com/go-logr/logr"
 
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
@@ -62,7 +63,7 @@ var createAclString = "User:%s,Topic,%s,%s,Create,Allow,*"
 // writeAclString is the raw representation of an ACL allowing Write on a Topic
 var writeAclString = "User:%s,Topic,%s,%s,Write,Allow,*"
 
-// reacAclString is the raw representation of an ACL allowing Read on a Topic
+// readAclString is the raw representation of an ACL allowing Read on a Topic
 var readAclString = "User:%s,Topic,%s,%s,Read,Allow,*"
 
 // readGroupAclString is the raw representation of an ACL allowing Read on ConsumerGroups
@@ -103,74 +104,65 @@ func GrantsToACLStrings(dn string, grants []v1alpha1.UserTopicGrant) []string {
 	return acls
 }
 
-func ShouldRefreshOnlyPerBrokerConfigs(currentConfigs, desiredConfigs map[string]string, log logr.Logger) bool {
-	touchedConfigs := collectTouchedConfigs(currentConfigs, desiredConfigs, log)
+func ShouldRefreshOnlyPerBrokerConfigs(currentConfigs, desiredConfigs *properties.Properties, log logr.Logger) bool {
+	// Get the diff of the configuration
+	configDiff := currentConfigs.Diff(desiredConfigs)
 
-	if _, ok := touchedConfigs[securityProtocolMapConfigName]; ok {
-		if listenersSecurityProtocolChanged(currentConfigs, desiredConfigs) {
+	// Return if there is no drift in the configuration
+	if len(configDiff) == 0 {
+		return true
+	}
+
+	log.V(1).Info("configs have been changed", "configs", configDiff)
+
+	if diff, ok := configDiff[securityProtocolMapConfigName]; ok {
+		if listenersSecurityProtocolChanged(diff[0].Value(), diff[1].Value()) {
 			return false
 		}
 	}
 
 	for _, perBrokerConfig := range PerBrokerConfigs {
-		delete(touchedConfigs, perBrokerConfig)
+		delete(configDiff, perBrokerConfig)
 	}
 
-	return len(touchedConfigs) == 0
+	return len(configDiff) == 0
 }
 
 // Security protocol cannot be updated for existing listener
 // a rolling upgrade should be triggered in this case
-func listenersSecurityProtocolChanged(currentConfigs, desiredConfigs map[string]string) bool {
+func listenersSecurityProtocolChanged(current, desired string) bool {
 	// added or deleted config is ok
-	if currentConfigs[securityProtocolMapConfigName] == "" || desiredConfigs[securityProtocolMapConfigName] == "" {
+	if current == "" || desired == "" {
 		return false
 	}
-	currentListenerProtocolMap := make(map[string]string)
-	for _, listenerConfig := range strings.Split(currentConfigs[securityProtocolMapConfigName], ",") {
-		listenerProtocol := strings.Split(listenerConfig, ":")
-		if len(listenerProtocol) != 2 {
-			continue
-		}
-		currentListenerProtocolMap[strings.TrimSpace(listenerProtocol[0])] = strings.TrimSpace(listenerProtocol[1])
+	currentConfig := newListenerSecurityProtocolMap(current)
+	desiredConfig := newListenerSecurityProtocolMap(desired)
+
+	if len(currentConfig) != len(desiredConfig) {
+		return true
 	}
-	for _, listenerConfig := range strings.Split(desiredConfigs[securityProtocolMapConfigName], ",") {
-		desiredListenerProtocol := strings.Split(listenerConfig, ":")
-		if len(desiredListenerProtocol) != 2 {
-			continue
-		}
-		if currentListenerProtocolValue, ok := currentListenerProtocolMap[strings.TrimSpace(desiredListenerProtocol[0])]; ok {
-			if currentListenerProtocolValue != strings.TrimSpace(desiredListenerProtocol[1]) {
-				return true
-			}
+
+	for dKey, dConf := range desiredConfig {
+		if cConf, ok := currentConfig[dKey]; ok && cConf != dConf {
+			return true
 		}
 	}
 	return false
 }
 
-// collects are the config keys that are either added, removed or updated
-// between the current and the desired ConfigMap
-func collectTouchedConfigs(currentConfigs, desiredConfigs map[string]string, log logr.Logger) map[string]struct{} {
-	touchedConfigs := make(map[string]struct{})
+type listenerSecurityProtocolMap map[string]string
 
-	currentConfigsCopy := make(map[string]string)
-	for k, v := range currentConfigs {
-		currentConfigsCopy[k] = v
-	}
+func newListenerSecurityProtocolMap(s string) listenerSecurityProtocolMap {
+	listenerSecProtoMap := make(listenerSecurityProtocolMap)
 
-	for configName, desiredValue := range desiredConfigs {
-		if currentValue, ok := currentConfigsCopy[configName]; !ok || currentValue != desiredValue {
-			// new or updated config
-			touchedConfigs[configName] = struct{}{}
+	for _, listenerConfig := range strings.Split(s, ",") {
+		listenerProto := strings.SplitN(listenerConfig, ":", 2)
+		// listenerProto must have 2 parts and it is considered as invalid if it does not.
+		if len(listenerProto) != 2 {
+			continue
 		}
-		delete(currentConfigsCopy, configName)
-	}
-
-	for configName := range currentConfigsCopy {
-		// deleted config
-		touchedConfigs[configName] = struct{}{}
+		listenerSecProtoMap[strings.TrimSpace(listenerProto[0])] = strings.TrimSpace(listenerProto[1])
 	}
 
-	log.V(1).Info("configs have been changed", "configs", touchedConfigs)
-	return touchedConfigs
+	return listenerSecProtoMap
 }
diff --git a/pkg/util/util.go b/pkg/util/util.go
index f91e83ab..51a7d24d 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"emperror.dev/errors"
+	"github.com/Shopify/sarama"
 	"github.com/go-logr/logr"
 	"github.com/imdario/mergo"
 	"go.uber.org/zap"
@@ -33,6 +34,7 @@ import (
 	k8s_zap "sigs.k8s.io/controller-runtime/pkg/log/zap"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
 const (
@@ -128,12 +130,14 @@ func IsSSLEnabledForInternalCommunication(l []v1beta1.InternalListenerConfig) (e
 	return enabled
 }
 
-// ConvertMapStringToMapStringPointer converts a simple map[string]string to map[string]*string
-func ConvertMapStringToMapStringPointer(inputMap map[string]string) map[string]*string {
+// ConvertPropertiesToMapStringPointer converts a Properties object to map[string]*string
+func ConvertPropertiesToMapStringPointer(pp *properties.Properties) map[string]*string {
 
-	result := map[string]*string{}
-	for key, value := range inputMap {
-		result[key] = StringPointer(value)
+	result := make(map[string]*string, pp.Len())
+	for _, key := range pp.Keys() {
+		if p, ok := pp.Get(key); ok {
+			result[key] = StringPointer(p.Value())
+		}
 	}
 	return result
 }
@@ -158,21 +162,6 @@ func StringSliceRemove(list []string, s string) []string {
 	return list
 }
 
-// ParsePropertiesFormat parses the properties format configuration into map[string]string
-func ParsePropertiesFormat(properties string) map[string]string {
-	kafkaProperties := strings.Split(strings.TrimSpace(properties), "\n")
-	config := make(map[string]string, len(kafkaProperties))
-
-	for _, prop := range kafkaProperties {
-		kvSeparatorIdx := strings.Index(prop, "=")
-		if kvSeparatorIdx >= 0 {
-			config[strings.TrimSpace(prop[:kvSeparatorIdx])] = strings.TrimSpace(prop[kvSeparatorIdx+1:])
-		}
-	}
-
-	return config
-}
-
 func AreStringSlicesIdentical(a, b []string) bool {
 	sort.Strings(a)
 	sort.Strings(b)
@@ -320,3 +309,17 @@ func CreateLogger(debug bool, development bool) logr.Logger {
 
 	return k8s_zap.New(k8s_zap.UseDevMode(development), k8s_zap.Encoder(encoder), k8s_zap.Level(level))
 }
+
+// ConvertConfigEntryListToProperties function takes []sarama.ConfigEntry and coverts it to Properties object
+func ConvertConfigEntryListToProperties(config []sarama.ConfigEntry) (*properties.Properties, error) {
+	p := properties.NewProperties()
+
+	for _, c := range config {
+		err := p.Set(c.Name, c.Value)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return p, nil
+}
diff --git a/properties/pkg/encode.go b/properties/pkg/encode.go
index a9a9335f..26b97b6b 100644
--- a/properties/pkg/encode.go
+++ b/properties/pkg/encode.go
@@ -96,7 +96,7 @@ func marshal(v interface{}) (*Properties, error) {
 		if vFieldValue.IsZero() && st.OmitEmpty {
 			continue
 		}
-		err = properties.Set(st.Key, vFieldValue.Interface(), "")
+		err = properties.Set(st.Key, vFieldValue.Interface())
 		if err != nil {
 			return nil, errors.WithDetails(err, "key", st.Key, "value", vFieldValue.Interface())
 		}
diff --git a/properties/pkg/properties.go b/properties/pkg/properties.go
index 0bf1e4f4..ec4ab2d5 100644
--- a/properties/pkg/properties.go
+++ b/properties/pkg/properties.go
@@ -119,7 +119,12 @@ func (p *Properties) Len() int {
 }
 
 // Add a new Property to Properties with the given name and value.
-func (p *Properties) Set(key string, value interface{}, comment string) error {
+func (p *Properties) Set(key string, value interface{}) error {
+	return p.SetWithComment(key, value, "")
+}
+
+// Add a new Property to Properties with the given name and value.
+func (p *Properties) SetWithComment(key string, value interface{}, comment string) error {
 	prop := Property{}
 	err := prop.set(key, value, comment)
 	if err != nil {
@@ -328,8 +333,8 @@ func (p *Properties) Diff(m *Properties) DiffResult {
 		keys[mKey] = struct{}{}
 	}
 
-	// Get difference of the tow Properties by value
-	diff := make(DiffResult, len(keys))
+	// Get difference of the two Properties by value
+	diff := make(DiffResult)
 
 	for key := range keys {
 		pProp, pFound := p.properties[key]


Test patch:
diff --git a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
index 0c777c8f..45b5bdc1 100644
--- a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
+++ b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
@@ -110,10 +110,8 @@ func expectCruiseControlConfigMap(kafkaCluster *v1beta1.KafkaCluster) {
 	Expect(configMap.Labels).To(HaveKeyWithValue("app", "cruisecontrol"))
 	Expect(configMap.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
 
-	Expect(configMap.Data).To(HaveKeyWithValue("cruisecontrol.properties", fmt.Sprintf(`some.config=value
-# The Kafka cluster to control.
-bootstrap.servers=%s-all-broker:29092
-# The zookeeper connect of the Kafka cluster
+	Expect(configMap.Data).To(HaveKeyWithValue("cruisecontrol.properties", fmt.Sprintf(`bootstrap.servers=%s-all-broker:29092
+some.config=value
 zookeeper.connect=/
 `, kafkaCluster.Name)))
 	Expect(configMap.Data).To(HaveKeyWithValue("capacity.json", `{
diff --git a/controllers/tests/kafkacluster_controller_kafka_test.go b/controllers/tests/kafkacluster_controller_kafka_test.go
index ecfca606..60a6311a 100644
--- a/controllers/tests/kafkacluster_controller_kafka_test.go
+++ b/controllers/tests/kafkacluster_controller_kafka_test.go
@@ -172,7 +172,8 @@ listener.security.protocol.map=INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,TEST:
 listeners=INTERNAL://:29092,CONTROLLER://:29093,TEST://:9094
 log.dirs=/kafka-logs/kafka
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
-zookeeper.connect=/`, 19090+broker.Id, broker.Id, broker.Id, broker.Id, broker.Id)))
+zookeeper.connect=/
+`, 19090+broker.Id, broker.Id, broker.Id, broker.Id, broker.Id)))
 
 	// assert log4j?
 }
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 04bf7fab..2fb41186 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -17,6 +17,7 @@ package kafka
 import (
 	"testing"
 
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -253,8 +254,18 @@ zookeeper.connect=example.zk:2181/`,
 
 			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, map[string]v1beta1.ListenerStatusList{}, map[string]v1beta1.ListenerStatusList{}, controllerListenerStatus, "", "", []string{}, logf.NullLogger{})
 
-			if generatedConfig != test.expectedConfig {
-				t.Errorf("the expected config is %s, received: %s", test.expectedConfig, generatedConfig)
+			generated, err := properties.NewFromString(generatedConfig)
+			if err != nil {
+				t.Fatalf("failed parsing generated configuration as Properties: %s", generatedConfig)
+			}
+
+			expected, err := properties.NewFromString(test.expectedConfig)
+			if err != nil {
+				t.Fatalf("failed parsing expected configuration as Properties: %s", expected)
+			}
+
+			if !expected.Equal(generated) {
+				t.Errorf("the expected config is:\n%s\nreceived:\n%s\n", test.expectedConfig, generatedConfig)
 			}
 		})
 	}
diff --git a/pkg/resources/kafka/configs_test.go b/pkg/resources/kafka/configs_test.go
index 4dac50e3..2956b99e 100644
--- a/pkg/resources/kafka/configs_test.go
+++ b/pkg/resources/kafka/configs_test.go
@@ -18,13 +18,14 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
 func TestCollectNonIdenticalConfigsFromResponse(t *testing.T) {
 	testCases := []struct {
 		Description  string
 		Response     []*sarama.ConfigEntry
-		BrokerConfig map[string]string
+		BrokerConfig string
 		Result       bool
 	}{
 		{
@@ -39,10 +40,9 @@ func TestCollectNonIdenticalConfigsFromResponse(t *testing.T) {
 					Value: "value2",
 				},
 			},
-			BrokerConfig: map[string]string{
-				"config1": "value1",
-				"config2": "value2",
-			},
+			BrokerConfig: `config1=value1
+config2=value2
+`,
 			Result: false,
 		},
 		{
@@ -61,10 +61,9 @@ func TestCollectNonIdenticalConfigsFromResponse(t *testing.T) {
 					Value: "value3",
 				},
 			},
-			BrokerConfig: map[string]string{
-				"config1": "value1",
-				"config2": "value2",
-			},
+			BrokerConfig: `config1=value1
+config2=value2
+`,
 			Result: false,
 		},
 		{
@@ -79,15 +78,18 @@ func TestCollectNonIdenticalConfigsFromResponse(t *testing.T) {
 					Value: "value2",
 				},
 			},
-			BrokerConfig: map[string]string{
-				"config1": "value1",
-				"config2": "value3",
-			},
+			BrokerConfig: `config1=value1
+config2=value3
+`,
 			Result: true,
 		},
 	}
 	for _, testCase := range testCases {
-		if shouldUpdatePerBrokerConfig(testCase.Response, testCase.BrokerConfig) != testCase.Result {
+		brokerConf, err := properties.NewFromString(testCase.BrokerConfig)
+		if err != nil {
+			t.Errorf("failed parsing Properties from string: %v\n", testCase.BrokerConfig)
+		}
+		if shouldUpdatePerBrokerConfig(testCase.Response, brokerConf) != testCase.Result {
 			t.Errorf("wrong result generated by test case: %s", testCase.Description)
 		}
 	}
diff --git a/pkg/util/kafka/common_test.go b/pkg/util/kafka/common_test.go
index e2f29650..3f509b2b 100644
--- a/pkg/util/kafka/common_test.go
+++ b/pkg/util/kafka/common_test.go
@@ -15,186 +15,90 @@
 package kafka
 
 import (
-	"reflect"
 	"testing"
 
-	corev1 "k8s.io/api/core/v1"
-
 	"github.com/banzaicloud/kafka-operator/pkg/util"
+	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
 
 func TestShouldRefreshOnlyPerBrokerConfigs(t *testing.T) {
 	testCases := []struct {
 		Description    string
-		CurrentConfigs map[string]string
-		DesiredConfigs map[string]string
+		CurrentConfigs string
+		DesiredConfigs string
 		Result         bool
 	}{
 		{
 			Description: "configs did not change",
-			CurrentConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"unmodified_config_2": "unmodified_value_2",
-			},
-			DesiredConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"unmodified_config_2": "unmodified_value_2",
-			},
+			CurrentConfigs: `unmodified_config_1=unmodified_value_1
+unmodified_config_2=unmodified_value_2
+`,
+			DesiredConfigs: `unmodified_config_1=unmodified_value_1
+unmodified_config_2=unmodified_value_2
+`,
 			Result: true,
 		},
 		{
 			Description: "only non per-broker config changed",
-			CurrentConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"unmodified_config_2": "modified_value_2",
-			},
-			DesiredConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"unmodified_config_2": "modified_value_3",
-			},
+			CurrentConfigs: `unmodified_config_1=unmodified_value_1
+unmodified_config_2=modified_value_2
+`,
+			DesiredConfigs: `unmodified_config_1=unmodified_value_1
+unmodified_config_2=modified_value_3
+`,
 			Result: false,
 		},
 		{
 			Description: "only per-broker config changed",
-			CurrentConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"ssl.client.auth":     "modified_value_2",
-			},
-			DesiredConfigs: map[string]string{
-				"unmodified_config_1": "unmodified_value_1",
-				"ssl.client.auth":     "modified_value_3",
-			},
+			CurrentConfigs: `unmodified_config_1=unmodified_value_1
+ssl.client.auth=modified_value_2
+`,
+			DesiredConfigs: `unmodified_config_1=unmodified_value_1
+ssl.client.auth=modified_value_3
+`,
 			Result: true,
 		},
 		{
 			Description: "per-broker and non per-broker configs have changed",
-			CurrentConfigs: map[string]string{
-				"modified_config_1": "modified_value_1",
-				"ssl.client.auth":   "modified_value_3",
-			},
-			DesiredConfigs: map[string]string{
-				"modified_config_1": "modified_value_2",
-				"ssl.client.auth":   "modified_value_4",
-			},
+			CurrentConfigs: `modified_config_1=modified_value_1
+ssl.client.auth=modified_value_3
+`,
+			DesiredConfigs: `modified_config_1=modified_value_2
+ssl.client.auth=modified_value_4
+`,
 			Result: false,
 		},
 		{
-			Description: "security protocol map can be changed as a per-broker config",
-			CurrentConfigs: map[string]string{
-				"listener.security.protocol.map": "listener1:protocol1,listener2:protocol2",
-			},
-			DesiredConfigs: map[string]string{
-				"listener.security.protocol.map": "listener1:protocol1,listener3:protocol3",
-			},
-			Result: true,
+			Description:    "security protocol map can be changed as a per-broker config",
+			CurrentConfigs: "listener.security.protocol.map=listener1:protocol1,listener2:protocol2",
+			DesiredConfigs: "listener.security.protocol.map=listener1:protocol1,listener3:protocol3",
+			Result:         true,
 		},
 		{
-			Description: "security protocol map can't be changed as a per-broker config",
-			CurrentConfigs: map[string]string{
-				"listener.security.protocol.map": "listener1:protocol1,listener2:protocol2",
-			},
-			DesiredConfigs: map[string]string{
-				"listener.security.protocol.map": "listener1:protocol1,listener2:protocol3",
-			},
-			Result: false,
+			Description:    "security protocol map can't be changed as a per-broker config",
+			CurrentConfigs: "listener.security.protocol.map=listener1:protocol1,listener2:protocol2",
+			DesiredConfigs: "listener.security.protocol.map=listener1:protocol1,listener2:protocol3",
+			Result:         false,
 		},
 		{
 			Description:    "security protocol map added as config",
-			CurrentConfigs: map[string]string{},
-			DesiredConfigs: map[string]string{
-				"listener.security.protocol.map": "listener1:protocol1,listener2:protocol2",
-			},
-			Result: true,
+			CurrentConfigs: "",
+			DesiredConfigs: "listener.security.protocol.map=listener1:protocol1,listener2:protocol2",
+			Result:         true,
 		},
 	}
 	logger := util.CreateLogger(false, false)
 	for i, testCase := range testCases {
-		if ShouldRefreshOnlyPerBrokerConfigs(testCase.CurrentConfigs, testCase.DesiredConfigs, logger) != testCase.Result {
-			t.Errorf("test case %d failed: %s", i, testCase.Description)
+		current, err := properties.NewFromString(testCase.CurrentConfigs)
+		if err != nil {
+			t.Fatalf("failed to parse Properties from string: %s", testCase.CurrentConfigs)
 		}
-	}
-}
-
-func TestCollectTouchedConfigs(t *testing.T) {
-	testCases := []struct {
-		DesiredConfigs string
-		CurrentConfigs string
-		Result         map[string]struct{}
-	}{
-		{
-			DesiredConfigs: `key1=value1
-key2=value2
-key3=value3`,
-			CurrentConfigs: `key1=value1
-key2=value2
-key3=value3`,
-			Result: map[string]struct{}{},
-		},
-		{
-			DesiredConfigs: `key1=value1
-key2=value2`,
-			CurrentConfigs: `key1=value1
-key2=value2
-key3=value3`,
-			Result: map[string]struct{}{
-				"key3": {},
-			},
-		},
-		{
-			DesiredConfigs: `key1=value1
-key2=value2
-key3=value3`,
-			CurrentConfigs: `key1=value1
-key2=value2`,
-			Result: map[string]struct{}{
-				"key3": {},
-			},
-		},
-		{
-			DesiredConfigs: `key1=value1
-key2=value2
-key3=value3`,
-			CurrentConfigs: `key1=value1
-key2=value4
-key3=value3`,
-			Result: map[string]struct{}{
-				"key2": {},
-			},
-		},
-		{
-			DesiredConfigs: `key1=value1
-key2=value2
-key3=value3
-key4=value4`,
-			CurrentConfigs: `key1=value6
-key2=value2
-key3=value3
-key5=value5`,
-			Result: map[string]struct{}{
-				"key1": {},
-				"key4": {},
-				"key5": {},
-			},
-		},
-	}
-
-	logger := util.CreateLogger(false, false)
-	for _, testCase := range testCases {
-		desired := &corev1.ConfigMap{
-			Data: map[string]string{
-				ConfigPropertyName: testCase.DesiredConfigs,
-			},
-		}
-		current := &corev1.ConfigMap{
-			Data: map[string]string{
-				ConfigPropertyName: testCase.CurrentConfigs,
-			},
+		desired, err := properties.NewFromString(testCase.DesiredConfigs)
+		if err != nil {
+			t.Fatalf("failed to parse Properties from string: %s", testCase.DesiredConfigs)
 		}
-		touchedConfigs := collectTouchedConfigs(
-			util.ParsePropertiesFormat(current.Data[ConfigPropertyName]),
-			util.ParsePropertiesFormat(desired.Data[ConfigPropertyName]), logger)
-		if !reflect.DeepEqual(touchedConfigs, testCase.Result) {
-			t.Errorf("comparison failed - expected: %s, actual: %s", testCase.Result, touchedConfigs)
+		if ShouldRefreshOnlyPerBrokerConfigs(current, desired, logger) != testCase.Result {
+			t.Errorf("test case %d failed: %s", i, testCase.Description)
 		}
 	}
 }
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 3376b2aa..6aeade11 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -387,23 +387,6 @@ func TestGetBrokerConfigUniqueStorage(t *testing.T) {
 	}
 }
 
-func TestParsePropertiesFormat(t *testing.T) {
-	testProp := `
-broker.id=1
-advertised.listener=broker-1:29092
-empty.config=
- trailing.whitespace=foobar
-`
-	props := ParsePropertiesFormat(testProp)
-
-	if props["broker.id"] != "1" ||
-		props["advertised.listener"] != "broker-1:29092" ||
-		props["empty.config"] != "" ||
-		props["trailing.whitespace"] != "foobar" {
-		t.Error("Error properties not loaded correctly")
-	}
-}
-
 func TestIntstrPointer(t *testing.T) {
 	i := int(10)
 	ptr := IntstrPointer(i)
diff --git a/properties/pkg/decode_test.go b/properties/pkg/decode_test.go
index 7d556457..0165634a 100644
--- a/properties/pkg/decode_test.go
+++ b/properties/pkg/decode_test.go
@@ -90,11 +90,11 @@ func TestUnmarshal(t *testing.T) {
 		expectedList := []string{"test value1", "test value2"}
 
 		p := NewProperties()
-		_ = p.Set("string.field", expectedString, "")
-		_ = p.Set("int.field", expectedInt, "")
-		_ = p.Set("bool.field", expectedBool, "")
-		_ = p.Set("float.field", expectedFloat, "")
-		_ = p.Set("list.field", expectedList, "")
+		_ = p.Set("string.field", expectedString)
+		_ = p.Set("int.field", expectedInt)
+		_ = p.Set("bool.field", expectedBool)
+		_ = p.Set("float.field", expectedFloat)
+		_ = p.Set("list.field", expectedList)
 
 		err := Unmarshal(p, s)
 
@@ -130,7 +130,7 @@ func TestUnmarshal(t *testing.T) {
 
 		expectedValue := "after unmarshal"
 		p := NewProperties()
-		_ = p.Set("custom.field", expectedValue, "")
+		_ = p.Set("custom.field", expectedValue)
 
 		err := Unmarshal(p, s)
 
diff --git a/properties/pkg/encode_test.go b/properties/pkg/encode_test.go
index c0ac7b4d..ea616b66 100644
--- a/properties/pkg/encode_test.go
+++ b/properties/pkg/encode_test.go
@@ -50,7 +50,7 @@ type TestMarshalerStruct struct {
 
 func (s TestMarshalerStruct) MarshalProperties() (*Properties, error) {
 	p := NewProperties()
-	_ = p.Set("custom.marshaller.called", "true", "")
+	_ = p.Set("custom.marshaller.called", "true")
 	return p, nil
 }
 
@@ -109,11 +109,11 @@ func TestMarshal(t *testing.T) {
 		}
 
 		expected := NewProperties()
-		_ = expected.Set("string.field", "property string", "")
-		_ = expected.Set("int.field", 100, "")
-		_ = expected.Set("bool.field", true, "")
-		_ = expected.Set("float.field", 128.9, "")
-		_ = expected.Set("list.field", []string{"test value1", "test value2"}, "")
+		_ = expected.Set("string.field", "property string")
+		_ = expected.Set("int.field", 100)
+		_ = expected.Set("bool.field", true)
+		_ = expected.Set("float.field", 128.9)
+		_ = expected.Set("list.field", []string{"test value1", "test value2"})
 
 		if !p.Equal(expected) {
 			t.Errorf("Mismatch in expected and returned Properties!\nExpected: %q\nGot: %q\n", expected, p)
@@ -139,12 +139,12 @@ func TestMarshal(t *testing.T) {
 		}
 
 		expected := NewProperties()
-		_ = expected.Set("string.field", "property string", "")
-		_ = expected.Set("int.field", 100, "")
-		_ = expected.Set("bool.field", true, "")
-		_ = expected.Set("float.field", 128.9, "")
-		_ = expected.Set("list.field", "test value1,test value2", "")
-		_ = expected.Set("omitempty.field", "omitempty property field", "")
+		_ = expected.Set("string.field", "property string")
+		_ = expected.Set("int.field", 100)
+		_ = expected.Set("bool.field", true)
+		_ = expected.Set("float.field", 128.9)
+		_ = expected.Set("list.field", "test value1,test value2")
+		_ = expected.Set("omitempty.field", "omitempty property field")
 
 		if !cmp.Equal(p, expected, cmp.AllowUnexported(Properties{}), cmp.AllowUnexported(Property{})) {
 			t.Errorf("Mismatch in expected and returned Properties!\nExpected: %q\nGot: %q\n\n %v\n", expected, p, cmp.Diff(expected, p))
@@ -177,7 +177,7 @@ func TestMarshal(t *testing.T) {
 		}
 
 		expected := NewProperties()
-		_ = expected.Set("custom.marshaller.called", "true", "")
+		_ = expected.Set("custom.marshaller.called", "true")
 
 		if !cmp.Equal(p, expected, cmp.AllowUnexported(Properties{}), cmp.AllowUnexported(Property{})) {
 			t.Errorf("Mismatch in expected and returned Properties!\nExpected: %q\nGot: %q\n\n %v\n", expected, p, cmp.Diff(expected, p))
diff --git a/properties/pkg/properties_test.go b/properties/pkg/properties_test.go
index 355e3473..e97a6e54 100644
--- a/properties/pkg/properties_test.go
+++ b/properties/pkg/properties_test.go
@@ -114,6 +114,80 @@ func TestProperties_Set(t *testing.T) {
 
 	p := NewProperties()
 
+	t.Run("Add new Property with string value to Properties", func(t *testing.T) {
+		expectedProperty := Property{
+			key:   "test.key3",
+			value: "test.value3",
+		}
+
+		err := p.Set(expectedProperty.key, expectedProperty.value)
+
+		if err != nil {
+			t.Error("Adding a new Property should not result an error.")
+		}
+
+		prop, found := p.Get(expectedProperty.key)
+
+		if !found {
+			t.Error("Newly added Property is not found.")
+		}
+
+		if !prop.Equal(expectedProperty) {
+			t.Errorf("Mismatch between return and expected list of keys. Expected: %q, got %q", prop, expectedProperty)
+		}
+	})
+
+	t.Run("Add new Property with integer value to Properties", func(t *testing.T) {
+		expectedProperty := Property{
+			key:   "test.key3",
+			value: "100",
+		}
+
+		err := p.Set(expectedProperty.key, 100)
+
+		if err != nil {
+			t.Error("Adding a new property should not result an error.")
+		}
+
+		prop, found := p.Get(expectedProperty.key)
+
+		if !found {
+			t.Errorf("Newly added Property is not found.")
+		}
+
+		if !prop.Equal(expectedProperty) {
+			t.Errorf("Mismatch between return and expected list of keys. Expected: %v, got %v", prop, expectedProperty)
+		}
+	})
+
+	t.Run("Add new Property with string slice value to Properties", func(t *testing.T) {
+		expectedProperty := Property{
+			key:   "test.key3",
+			value: "value1,value2,value3",
+		}
+
+		err := p.Set(expectedProperty.key, []string{"value1", "value2", "value3"})
+
+		if err != nil {
+			t.Error("Adding a new property should not result an error.")
+		}
+
+		prop, found := p.Get(expectedProperty.key)
+
+		if !found {
+			t.Errorf("Newly added Property is not found.")
+		}
+
+		if !prop.Equal(expectedProperty) {
+			t.Errorf("Mismatch between return and expected list of keys. Expected: %v, got %v", prop, expectedProperty)
+		}
+	})
+}
+
+func TestProperties_SetWithComment(t *testing.T) {
+
+	p := NewProperties()
+
 	t.Run("Add new Property with string value to Properties", func(t *testing.T) {
 		expectedProperty := Property{
 			key:     "test.key3",
@@ -121,7 +195,7 @@ func TestProperties_Set(t *testing.T) {
 			comment: "this is a comment line",
 		}
 
-		err := p.Set(expectedProperty.key, expectedProperty.value, expectedProperty.comment)
+		err := p.SetWithComment(expectedProperty.key, expectedProperty.value, expectedProperty.comment)
 
 		if err != nil {
 			t.Error("Adding a new Property should not result an error.")
@@ -140,11 +214,36 @@ func TestProperties_Set(t *testing.T) {
 
 	t.Run("Add new Property with integer value to Properties", func(t *testing.T) {
 		expectedProperty := Property{
-			key:   "test.key3",
-			value: "100",
+			key:     "test.key3",
+			value:   "100",
+			comment: "this is a comment line",
+		}
+
+		err := p.SetWithComment(expectedProperty.key, 100, expectedProperty.comment)
+
+		if err != nil {
+			t.Error("Adding a new property should not result an error.")
+		}
+
+		prop, found := p.Get(expectedProperty.key)
+
+		if !found {
+			t.Errorf("Newly added Property is not found.")
+		}
+
+		if !prop.Equal(expectedProperty) {
+			t.Errorf("Mismatch between return and expected list of keys. Expected: %v, got %v", prop, expectedProperty)
+		}
+	})
+
+	t.Run("Add new Property with string slice value to Properties", func(t *testing.T) {
+		expectedProperty := Property{
+			key:     "test.key3",
+			value:   "value1,value2,value3",
+			comment: "this is a comment line",
 		}
 
-		err := p.Set(expectedProperty.key, 100, "")
+		err := p.SetWithComment(expectedProperty.key, []string{"value1", "value2", "value3"}, expectedProperty.comment)
 
 		if err != nil {
 			t.Error("Adding a new property should not result an error.")