Repository: banzaicloud/kafka-operator
Pull request #602 (2021-05-04 09:12:04)

Move GetBrokerConfig utility function to the API
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | no
| API breaks?     | no
| Deprecations?   | no
| Related tickets | -
| License         | Apache 2.0

### What's in this PR?
<!-- Explain the contents of the PR. Give an overview about the implementation, which decisions were made and why. -->
Migrated the `util.GetBrokerConfig()` function to the API.

### Why?
<!-- Which problem does the PR fix? (Please remove this section if you linked an issue above) -->
Dependent packages could use that function instead of implementing the same logic that is in the operator pacakge.

### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [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


Patch:
diff --git a/internal/alertmanager/currentalert/process.go b/internal/alertmanager/currentalert/process.go
index b03b300c..05fe96f6 100644
--- a/internal/alertmanager/currentalert/process.go
+++ b/internal/alertmanager/currentalert/process.go
@@ -253,7 +253,7 @@ func resizePvc(log logr.Logger, labels model.LabelSet, annotiations model.LabelS
 
 	for i, broker := range cr.Spec.Brokers {
 		if strconv.Itoa(int(broker.Id)) == pvc.Labels["brokerId"] {
-			brokerConfig, err := util.GetBrokerConfig(broker, cr.Spec)
+			brokerConfig, err := broker.GetBrokerConfig(cr.Spec.BrokerConfigGroups)
 			if err != nil {
 				return errors.WrapIf(err, "failed to determine broker config")
 			}
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index b1e846e9..0827e07c 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -235,7 +235,7 @@ func generateDefaultBrokerCapacity() BrokerCapacity {
 }
 
 func generateBrokerNetworkIn(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := util.GetBrokerConfig(broker, kafkaClusterSpec)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
 	if err != nil {
 		log.V(warnLevel).Info("could not get incoming network resource limits falling back to default value")
 		return storageConfigNWINDefaultValue
@@ -249,7 +249,7 @@ func generateBrokerNetworkIn(broker v1beta1.Broker, kafkaClusterSpec v1beta1.Kaf
 }
 
 func generateBrokerNetworkOut(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := util.GetBrokerConfig(broker, kafkaClusterSpec)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
 	if err != nil {
 		log.V(warnLevel).Info("could not get outgoing network resource limits falling back to default value")
 		return storageConfigNWOUTDefaultValue
@@ -263,7 +263,7 @@ func generateBrokerNetworkOut(broker v1beta1.Broker, kafkaClusterSpec v1beta1.Ka
 }
 
 func generateBrokerCPU(broker v1beta1.Broker, kafkaClusterSpec v1beta1.KafkaClusterSpec, log logr.Logger) string {
-	brokerConfig, err := util.GetBrokerConfig(broker, kafkaClusterSpec)
+	brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
 	if err != nil {
 		log.V(warnLevel).Info("could not get cpu resource limits falling back to default value")
 		return storageConfigCPUDefaultValue
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 43b9c0e1..e5c2bb5b 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -254,8 +254,25 @@ func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger)
 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 {
+	finalBrokerConfig := getBrokerReadOnlyConfig(id, r.KafkaCluster, log)
+
+	// Get operator generated configuration
+	opGenConf := r.getConfigProperties(brokerConfig, id, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)
+
+	// Merge operator generated configuration to the final one
+	if opGenConf != nil {
+		finalBrokerConfig.Merge(opGenConf)
+	}
+
+	finalBrokerConfig.Sort()
+
+	return finalBrokerConfig.String()
+}
+
+// TODO move this into pkg/sdk in the future (adamantal)
+func getBrokerReadOnlyConfig(id int32, kafkaCluster *v1beta1.KafkaCluster, log logr.Logger) *properties.Properties {
 	// Parse cluster-wide readonly configuration
-	finalBrokerConfig, err := properties.NewFromString(r.KafkaCluster.Spec.ReadOnlyConfig)
+	finalBrokerConfig, err := properties.NewFromString(kafkaCluster.Spec.ReadOnlyConfig)
 	if err != nil {
 		log.Error(err, "failed to parse readonly cluster configuration")
 	}
@@ -263,7 +280,7 @@ func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerC
 	// Parse readonly broker configuration
 	var parsedReadOnlyBrokerConfig *properties.Properties
 	// Find configuration for broker with id
-	for _, broker := range r.KafkaCluster.Spec.Brokers {
+	for _, broker := range kafkaCluster.Spec.Brokers {
 		if broker.Id == id {
 			parsedReadOnlyBrokerConfig, err = properties.NewFromString(broker.ReadOnlyConfig)
 			if err != nil {
@@ -278,15 +295,5 @@ func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerC
 		finalBrokerConfig.Merge(parsedReadOnlyBrokerConfig)
 	}
 
-	// Get operator generated configuration
-	opGenConf := r.getConfigProperties(brokerConfig, id, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)
-
-	// Merge operator generated configuration to the final one
-	if opGenConf != nil {
-		finalBrokerConfig.Merge(opGenConf)
-	}
-
-	finalBrokerConfig.Sort()
-
-	return finalBrokerConfig.String()
+	return finalBrokerConfig
 }
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index ea5f06f0..3d2e270a 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -206,7 +206,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	brokersVolumes := make(map[string][]*corev1.PersistentVolumeClaim, len(r.KafkaCluster.Spec.Brokers))
 	for _, broker := range r.KafkaCluster.Spec.Brokers {
-		brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
 		if err != nil {
 			return errors.WrapIf(err, "failed to reconcile resource")
 		}
@@ -229,7 +229,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	reorderedBrokers := r.reorderBrokers(log, r.KafkaCluster.Spec.Brokers)
 	for _, broker := range reorderedBrokers {
-		brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
 		if err != nil {
 			return errors.WrapIf(err, "failed to reconcile resource")
 		}
@@ -879,7 +879,7 @@ func (r *Reconciler) createExternalListenerStatuses(log logr.Logger) (map[string
 				portNumber := eListener.ExternalStartingPort + broker.Id
 
 				if eListener.GetAccessMethod() != corev1.ServiceTypeLoadBalancer {
-					bConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+					bConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
 					if err != nil {
 						return nil, err
 					}
@@ -902,7 +902,7 @@ func (r *Reconciler) createExternalListenerStatuses(log logr.Logger) (map[string
 						brokerHost = fmt.Sprintf("%s-%d-%s.%s%s", r.KafkaCluster.Name, broker.Id, eListener.Name, r.KafkaCluster.Namespace, brokerHost)
 					}
 				}
-				brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+				brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
 				if err != nil {
 					return nil, err
 				}
diff --git a/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go b/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
index 2a0f3a66..87a962cd 100644
--- a/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
+++ b/pkg/resources/nodeportexternalaccess/nodeportExternalAccess.go
@@ -22,7 +22,6 @@ import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
 	"github.com/banzaicloud/kafka-operator/pkg/resources"
-	"github.com/banzaicloud/kafka-operator/pkg/util"
 )
 
 const (
@@ -53,7 +52,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
 			if eListener.GetAccessMethod() == corev1.ServiceTypeNodePort {
 				for _, broker := range r.KafkaCluster.Spec.Brokers {
-					brokerConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+					brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec.BrokerConfigGroups)
 					if err != nil {
 						return err
 					}
diff --git a/pkg/sdk/go.mod b/pkg/sdk/go.mod
index be213469..bdd98a1d 100644
--- a/pkg/sdk/go.mod
+++ b/pkg/sdk/go.mod
@@ -3,9 +3,12 @@ module github.com/banzaicloud/kafka-operator/pkg/sdk
 go 1.15
 
 require (
+	emperror.dev/errors v0.8.0
 	github.com/banzaicloud/istio-client-go v0.0.9
 	github.com/google/go-cmp v0.4.1 // indirect
+	github.com/imdario/mergo v0.3.11
 	github.com/jetstack/cert-manager v0.15.2
+	gotest.tools v2.2.0+incompatible
 	k8s.io/api v0.18.9
 	k8s.io/apimachinery v0.18.9
 	sigs.k8s.io/controller-runtime v0.6.3
diff --git a/pkg/sdk/go.sum b/pkg/sdk/go.sum
index 9abe0bb3..c3da5346 100644
--- a/pkg/sdk/go.sum
+++ b/pkg/sdk/go.sum
@@ -1,6 +1,8 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU=
+emperror.dev/errors v0.8.0 h1:4lycVEx0sdJkwDUfQ9pdu6SR0x7rgympt5f4+ok8jDk=
+emperror.dev/errors v0.8.0/go.mod h1:YcRvLPh626Ubn2xqtoprejnA5nFha+TJ+2vew48kWuE=
 github.com/Azure/azure-sdk-for-go v32.5.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=
 github.com/Azure/go-autorest/autorest v0.9.0/go.mod h1:xyHB1BMZT0cuDHU7I0+g046+BFDTQ8rEZB0s4Yfa6bI=
@@ -229,6 +231,8 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
 github.com/imdario/mergo v0.3.5/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.6/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
+github.com/imdario/mergo v0.3.11 h1:3tnifQM4i+fbajXKBHXWEH+KvNHqojZ778UH75j3bGA=
+github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jetstack/cert-manager v0.15.2 h1:3P2d0aV0j7hOb5/QK2tSwWHQITb/QQEizGqqdoq+lD4=
 github.com/jetstack/cert-manager v0.15.2/go.mod h1:7V2UW1EzgIWVUWi4uVATMIWXqinFOEqpggdvFdNMhlk=
@@ -319,6 +323,8 @@ github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndrE9hABlRI=
@@ -381,7 +387,11 @@ go.mongodb.org/mongo-driver v1.1.2/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qL
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
 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.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.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
+go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -539,6 +549,7 @@ gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20190905181640-827449938966/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
diff --git a/pkg/sdk/v1beta1/kafkacluster_types.go b/pkg/sdk/v1beta1/kafkacluster_types.go
index c8b725b5..9d63657c 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types.go
@@ -17,6 +17,9 @@ package v1beta1
 import (
 	"strings"
 
+	"emperror.dev/errors"
+	"github.com/imdario/mergo"
+
 	"github.com/banzaicloud/istio-client-go/pkg/networking/v1alpha3"
 
 	cmmeta "github.com/jetstack/cert-manager/pkg/apis/meta/v1"
@@ -1055,6 +1058,62 @@ func (mConfig *MonitoringConfig) GetCCJMXExporterConfig() string {
 `
 }
 
+// GetBrokerConfig composes the brokerConfig for a given broker using the broker's config group
+func (b *Broker) GetBrokerConfig(brokerConfigGroups map[string]BrokerConfig) (*BrokerConfig, error) {
+	bConfig := &BrokerConfig{}
+	if b.BrokerConfigGroup == "" {
+		return b.BrokerConfig, nil
+	} else if b.BrokerConfig != nil {
+		bConfig = b.BrokerConfig.DeepCopy()
+	}
+
+	groupConfig, exists := brokerConfigGroups[b.BrokerConfigGroup]
+	if !exists {
+		return nil, errors.NewWithDetails("missing brokerConfigGroup", "key", b.BrokerConfigGroup)
+	}
+
+	dstAffinity := &corev1.Affinity{}
+	srcAffinity := &corev1.Affinity{}
+
+	if groupConfig.Affinity != nil {
+		dstAffinity = groupConfig.Affinity.DeepCopy()
+	}
+	if bConfig.Affinity != nil {
+		srcAffinity = bConfig.Affinity
+	}
+
+	if err := mergo.Merge(dstAffinity, srcAffinity, mergo.WithOverride); err != nil {
+		return nil, errors.WrapIf(err, "could not merge brokerConfig.Affinity with ConfigGroup.Affinity")
+	}
+
+	err := mergo.Merge(bConfig, groupConfig, mergo.WithAppendSlice)
+	if err != nil {
+		return nil, errors.WrapIf(err, "could not merge brokerConfig with ConfigGroup")
+	}
+
+	bConfig.StorageConfigs = dedupStorageConfigs(bConfig.StorageConfigs)
+
+	if groupConfig.Affinity != nil || bConfig.Affinity != nil {
+		bConfig.Affinity = dstAffinity
+	}
+
+	return bConfig, nil
+}
+
+func dedupStorageConfigs(elements []StorageConfig) []StorageConfig {
+	encountered := make(map[string]struct{})
+	result := make([]StorageConfig, 0)
+
+	for _, v := range elements {
+		if _, ok := encountered[v.MountPath]; !ok {
+			encountered[v.MountPath] = struct{}{}
+			result = append(result, v)
+		}
+	}
+
+	return result
+}
+
 func cloneAnnotationMap(original map[string]string) map[string]string {
 	m := make(map[string]string, len(original))
 	for k, v := range original {
diff --git a/pkg/util/kafka/common.go b/pkg/util/kafka/common.go
index 935e48a8..30a42e77 100644
--- a/pkg/util/kafka/common.go
+++ b/pkg/util/kafka/common.go
@@ -245,7 +245,7 @@ func GatherBrokerConfigIfAvailable(kafkaClusterSpec v1beta1.KafkaClusterSpec, br
 	// but the broker spec does not contain the required config values.
 	for _, broker := range kafkaClusterSpec.Brokers {
 		if int(broker.Id) == brokerID {
-			brokerConfig, err := util.GetBrokerConfig(broker, kafkaClusterSpec)
+			brokerConfig, err := broker.GetBrokerConfig(kafkaClusterSpec.BrokerConfigGroups)
 			if err != nil {
 				return nil, err
 			}
diff --git a/pkg/util/util.go b/pkg/util/util.go
index b440d1cd..14447f16 100644
--- a/pkg/util/util.go
+++ b/pkg/util/util.go
@@ -29,7 +29,6 @@ import (
 	"github.com/imdario/mergo"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
-	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	k8s_zap "sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -206,7 +205,7 @@ func IsIngressConfigInUse(iConfigName, defaultConfigName string, cluster *v1beta
 	}
 	// Check if the given iConfigName is bound to a given broker
 	for _, broker := range cluster.Spec.Brokers {
-		brokerConfig, err := GetBrokerConfig(broker, cluster.Spec)
+		brokerConfig, err := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
 		if err != nil {
 			log.Error(err, "could not determine if ingressConfig is in use")
 			return false
@@ -319,62 +318,6 @@ func GetIngressConfigs(kafkaClusterSpec v1beta1.KafkaClusterSpec,
 	return ingressConfigs, defaultIngressConfigName, nil
 }
 
-// GetBrokerConfig compose the brokerConfig for a given broker
-func GetBrokerConfig(broker v1beta1.Broker, clusterSpec v1beta1.KafkaClusterSpec) (*v1beta1.BrokerConfig, error) {
-	bConfig := &v1beta1.BrokerConfig{}
-	if broker.BrokerConfigGroup == "" {
-		return broker.BrokerConfig, nil
-	} else if broker.BrokerConfig != nil {
-		bConfig = broker.BrokerConfig.DeepCopy()
-	}
-
-	groupConfig, exists := clusterSpec.BrokerConfigGroups[broker.BrokerConfigGroup]
-	if !exists {
-		return nil, errors.NewWithDetails("missing brokerConfigGroup", "key", broker.BrokerConfigGroup)
-	}
-
-	dstAffinity := &corev1.Affinity{}
-	srcAffinity := &corev1.Affinity{}
-
-	if groupConfig.Affinity != nil {
-		dstAffinity = groupConfig.Affinity.DeepCopy()
-	}
-	if bConfig.Affinity != nil {
-		srcAffinity = bConfig.Affinity
-	}
-
-	if err := mergo.Merge(dstAffinity, srcAffinity, mergo.WithOverride); err != nil {
-		return nil, errors.WrapIf(err, "could not merge brokerConfig.Affinity with ConfigGroup.Affinity")
-	}
-
-	err := mergo.Merge(bConfig, groupConfig, mergo.WithAppendSlice)
-	if err != nil {
-		return nil, errors.WrapIf(err, "could not merge brokerConfig with ConfigGroup")
-	}
-
-	bConfig.StorageConfigs = dedupStorageConfigs(bConfig.StorageConfigs)
-
-	if groupConfig.Affinity != nil || bConfig.Affinity != nil {
-		bConfig.Affinity = dstAffinity
-	}
-
-	return bConfig, nil
-}
-
-func dedupStorageConfigs(elements []v1beta1.StorageConfig) []v1beta1.StorageConfig {
-	encountered := make(map[string]struct{})
-	result := make([]v1beta1.StorageConfig, 0)
-
-	for _, v := range elements {
-		if _, ok := encountered[v.MountPath]; !ok {
-			encountered[v.MountPath] = struct{}{}
-			result = append(result, v)
-		}
-	}
-
-	return result
-}
-
 // GetBrokerImage returns the used broker image
 func GetBrokerImage(brokerConfig *v1beta1.BrokerConfig, clusterImage string) string {
 	if brokerConfig.Image != "" {


Test patch:
diff --git a/pkg/resources/kafka/pod_test.go b/pkg/resources/kafka/pod_test.go
index 497223a8..e23adb20 100644
--- a/pkg/resources/kafka/pod_test.go
+++ b/pkg/resources/kafka/pod_test.go
@@ -22,7 +22,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
-	"github.com/banzaicloud/kafka-operator/pkg/util"
 )
 
 func TestGetAffinity(t *testing.T) {
@@ -64,18 +63,22 @@ func TestGetAffinity(t *testing.T) {
 	affinity := getAffinity(&nilAffinityBrokerConfig, &cluster)
 	assert.DeepEqual(t, affinity.PodAntiAffinity, defaultPodAntiAffinity.PodAntiAffinity)
 
-	mergedAffinityBrokerConfig, _ := util.GetBrokerConfig(v1beta1.Broker{
+	broker := v1beta1.Broker{
 		BrokerConfig: &nilAffinityBrokerConfig,
-	}, cluster.Spec)
+	}
+
+	mergedAffinityBrokerConfig, _ := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
 
 	// expecting old behavior
 	affinity = getAffinity(mergedAffinityBrokerConfig, &cluster)
 	assert.DeepEqual(t, affinity.PodAntiAffinity, defaultPodAntiAffinity.PodAntiAffinity)
 
-	mergedAffinityBrokerConfig2, _ := util.GetBrokerConfig(v1beta1.Broker{
+	broker = v1beta1.Broker{
 		BrokerConfigGroup: "test",
 		BrokerConfig:      &nilAffinityBrokerConfig,
-	}, cluster.Spec)
+	}
+
+	mergedAffinityBrokerConfig2, _ := broker.GetBrokerConfig(cluster.Spec.BrokerConfigGroups)
 
 	// expecting old behavior
 	affinity = getAffinity(mergedAffinityBrokerConfig2, &cluster)
diff --git a/pkg/sdk/v1beta1/kafkacluster_types_test.go b/pkg/sdk/v1beta1/kafkacluster_types_test.go
new file mode 100644
index 00000000..c85be9b4
--- /dev/null
+++ b/pkg/sdk/v1beta1/kafkacluster_types_test.go
@@ -0,0 +1,377 @@
+// Copyright © 2021 Banzai Cloud
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v1beta1
+
+import (
+	"reflect"
+	"testing"
+
+	"gotest.tools/assert"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// we expect the final result will be the two segment's union
+func TestGetBrokerConfigAffinityMergeBrokerNodeAffinityWithGroupsAntiAffinity(t *testing.T) {
+	expected := &corev1.Affinity{
+		NodeAffinity: &corev1.NodeAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+				NodeSelectorTerms: []corev1.NodeSelectorTerm{
+					{
+						MatchFields: []corev1.NodeSelectorRequirement{
+							{Key: "fruit", Operator: "in", Values: []string{"apple"}},
+						},
+					},
+				},
+			},
+			PreferredDuringSchedulingIgnoredDuringExecution: nil,
+		},
+		PodAntiAffinity: &corev1.PodAntiAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+				{
+					LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"}},
+					Namespaces:    nil,
+					TopologyKey:   "kubernetes.io/hostname",
+				},
+			},
+			PreferredDuringSchedulingIgnoredDuringExecution: nil,
+		},
+	}
+
+	brokerConfigGroups := map[string]BrokerConfig{
+		"default": {
+			Affinity: &corev1.Affinity{
+				NodeAffinity: &corev1.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+						NodeSelectorTerms: []corev1.NodeSelectorTerm{
+							{
+								MatchExpressions: nil,
+								MatchFields: []corev1.NodeSelectorRequirement{
+									{
+										Key:      "fruit",
+										Operator: "in",
+										Values:   []string{"apple"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &BrokerConfig{
+			Affinity: &corev1.Affinity{
+				PodAntiAffinity: &corev1.PodAntiAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+						{
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
+							},
+							TopologyKey: "kubernetes.io/hostname",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	if err != nil {
+		t.Error(err)
+	}
+	assert.DeepEqual(t, expected, config.Affinity)
+}
+
+// we expect the final result will be the two segment's union
+func TestGetBrokerConfigAffinityMergeEmptyBrokerConfigWithDefaultConfig(t *testing.T) {
+	expected := &corev1.Affinity{
+		NodeAffinity: &corev1.NodeAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+				NodeSelectorTerms: []corev1.NodeSelectorTerm{
+					{
+						MatchFields: []corev1.NodeSelectorRequirement{
+							{Key: "fruit", Operator: "in", Values: []string{"apple"}},
+						},
+					},
+				},
+			},
+			PreferredDuringSchedulingIgnoredDuringExecution: nil,
+		},
+		PodAntiAffinity: &corev1.PodAntiAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+				{
+					LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"}},
+					Namespaces:    nil,
+					TopologyKey:   "kubernetes.io/hostname",
+				},
+			},
+			PreferredDuringSchedulingIgnoredDuringExecution: nil,
+		},
+	}
+
+	brokerConfigGroups := map[string]BrokerConfig{
+		"default": {
+			Affinity: &corev1.Affinity{
+				PodAntiAffinity: &corev1.PodAntiAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+						{
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+							},
+							TopologyKey: "kubernetes.io/hostname",
+						},
+					},
+				},
+				NodeAffinity: &corev1.NodeAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+						NodeSelectorTerms: []corev1.NodeSelectorTerm{
+							{
+								MatchExpressions: nil,
+								MatchFields: []corev1.NodeSelectorRequirement{
+									{
+										Key:      "fruit",
+										Operator: "in",
+										Values:   []string{"apple"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+	}
+
+	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	if err != nil {
+		t.Error(err)
+	}
+
+	assert.DeepEqual(t, expected, config.Affinity)
+}
+
+// the equal values got duplicated
+func TestGetBrokerConfigAffinityMergeEqualPodAntiAffinity(t *testing.T) {
+	expected := &corev1.Affinity{
+		NodeAffinity: nil,
+		PodAffinity:  nil,
+		PodAntiAffinity: &corev1.PodAntiAffinity{
+			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+				{
+					LabelSelector: &metav1.LabelSelector{
+						MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
+					},
+					TopologyKey: "kubernetes.io/hostname",
+				},
+			},
+			PreferredDuringSchedulingIgnoredDuringExecution: nil,
+		},
+	}
+
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &BrokerConfig{
+			Affinity: &corev1.Affinity{
+				PodAntiAffinity: &corev1.PodAntiAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+						{
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
+							},
+							TopologyKey: "kubernetes.io/hostname",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	brokerConfigGroups := map[string]BrokerConfig{
+		"default": {
+			Affinity: &corev1.Affinity{
+				PodAntiAffinity: &corev1.PodAntiAffinity{
+					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+						{
+							LabelSelector: &metav1.LabelSelector{
+								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
+							},
+							TopologyKey: "kubernetes.io/hostname",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	config, err := broker.GetBrokerConfig(brokerConfigGroups)
+	if err != nil {
+		t.Error(err)
+	}
+	assert.DeepEqual(t, expected, config.Affinity)
+}
+
+func TestGetBrokerConfig(t *testing.T) {
+	expected := &BrokerConfig{
+		StorageConfigs: []StorageConfig{
+			{
+				MountPath: "kafka-test/log",
+				PvcSpec: &corev1.PersistentVolumeClaimSpec{
+					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+					Resources: corev1.ResourceRequirements{
+						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+					},
+				},
+			},
+			{
+				MountPath: "kafka-test1/log",
+				PvcSpec: &corev1.PersistentVolumeClaimSpec{
+					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+					Resources: corev1.ResourceRequirements{
+						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+					},
+				},
+			},
+		},
+	}
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &BrokerConfig{
+			StorageConfigs: []StorageConfig{
+				{
+					MountPath: "kafka-test/log",
+					PvcSpec: &corev1.PersistentVolumeClaimSpec{
+						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	brokerConfigGroups := map[string]BrokerConfig{
+		"default": {
+			StorageConfigs: []StorageConfig{
+				{
+					MountPath: "kafka-test1/log",
+					PvcSpec: &corev1.PersistentVolumeClaimSpec{
+						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	result, err := broker.GetBrokerConfig(brokerConfigGroups)
+	if err != nil {
+		t.Error("Error GetBrokerConfig throw an unexpected error")
+	}
+	if !reflect.DeepEqual(result, expected) {
+		t.Error("Expected:", expected, "Got:", result)
+	}
+}
+
+func TestGetBrokerConfigUniqueStorage(t *testing.T) {
+	expected := &BrokerConfig{
+		StorageConfigs: []StorageConfig{
+			{
+				MountPath: "kafka-test/log",
+				PvcSpec: &corev1.PersistentVolumeClaimSpec{
+					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+					Resources: corev1.ResourceRequirements{
+						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("20Gi")},
+					},
+				},
+			},
+			{
+				MountPath: "kafka-test1/log",
+				PvcSpec: &corev1.PersistentVolumeClaimSpec{
+					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+					Resources: corev1.ResourceRequirements{
+						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+					},
+				},
+			},
+		},
+	}
+	broker := Broker{
+		Id:                0,
+		BrokerConfigGroup: "default",
+		BrokerConfig: &BrokerConfig{
+			StorageConfigs: []StorageConfig{
+				{
+					MountPath: "kafka-test/log",
+					PvcSpec: &corev1.PersistentVolumeClaimSpec{
+						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("20Gi")},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	brokerConfigGroups := map[string]BrokerConfig{
+		"default": {
+			StorageConfigs: []StorageConfig{
+				{
+					MountPath: "kafka-test1/log",
+					PvcSpec: &corev1.PersistentVolumeClaimSpec{
+						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+						},
+					},
+				},
+				{
+					MountPath: "kafka-test/log",
+					PvcSpec: &corev1.PersistentVolumeClaimSpec{
+						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
+						Resources: corev1.ResourceRequirements{
+							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	result, err := broker.GetBrokerConfig(brokerConfigGroups)
+	if err != nil {
+		t.Error("Error GetBrokerConfig throw an unexpected error")
+	}
+	if !reflect.DeepEqual(result, expected) {
+		t.Error("Expected:", expected, "Got:", result)
+	}
+}
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index 42ebd5b9..a06cd702 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -24,371 +24,9 @@ import (
 	"gotest.tools/assert"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 )
 
-// we expect the final result will be the two segment's union
-func TestGetBrokerConfigAffinityMergeBrokerNodeAffinityWithGroupsAntiAffinity(t *testing.T) {
-	expected := &corev1.Affinity{
-		NodeAffinity: &corev1.NodeAffinity{
-			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-				NodeSelectorTerms: []corev1.NodeSelectorTerm{
-					{
-						MatchFields: []corev1.NodeSelectorRequirement{
-							{Key: "fruit", Operator: "in", Values: []string{"apple"}},
-						},
-					},
-				},
-			},
-			PreferredDuringSchedulingIgnoredDuringExecution: nil,
-		},
-		PodAntiAffinity: &corev1.PodAntiAffinity{
-			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-				{
-					LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"}},
-					Namespaces:    nil,
-					TopologyKey:   "kubernetes.io/hostname",
-				},
-			},
-			PreferredDuringSchedulingIgnoredDuringExecution: nil,
-		},
-	}
-
-	cluster := v1beta1.KafkaClusterSpec{
-		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
-			"default": {
-				Affinity: &corev1.Affinity{
-					NodeAffinity: &corev1.NodeAffinity{
-						RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-							NodeSelectorTerms: []corev1.NodeSelectorTerm{
-								{
-									MatchExpressions: nil,
-									MatchFields: []corev1.NodeSelectorRequirement{
-										{
-											Key:      "fruit",
-											Operator: "in",
-											Values:   []string{"apple"},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	broker := v1beta1.Broker{
-		Id:                0,
-		BrokerConfigGroup: "default",
-		BrokerConfig: &v1beta1.BrokerConfig{
-			Affinity: &corev1.Affinity{
-				PodAntiAffinity: &corev1.PodAntiAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-						{
-							LabelSelector: &metav1.LabelSelector{
-								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
-							},
-							TopologyKey: "kubernetes.io/hostname",
-						},
-					},
-				},
-			},
-		},
-	}
-
-	config, err := GetBrokerConfig(broker, cluster)
-	if err != nil {
-		t.Error(err)
-	}
-	assert.DeepEqual(t, expected, config.Affinity)
-}
-
-// we expect the final result will be the two segment's union
-func TestGetBrokerConfigAffinityMergeEmptyBrokerConfigWithDefaultConfig(t *testing.T) {
-	expected := &corev1.Affinity{
-		NodeAffinity: &corev1.NodeAffinity{
-			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-				NodeSelectorTerms: []corev1.NodeSelectorTerm{
-					{
-						MatchFields: []corev1.NodeSelectorRequirement{
-							{Key: "fruit", Operator: "in", Values: []string{"apple"}},
-						},
-					},
-				},
-			},
-			PreferredDuringSchedulingIgnoredDuringExecution: nil,
-		},
-		PodAntiAffinity: &corev1.PodAntiAffinity{
-			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-				{
-					LabelSelector: &metav1.LabelSelector{MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"}},
-					Namespaces:    nil,
-					TopologyKey:   "kubernetes.io/hostname",
-				},
-			},
-			PreferredDuringSchedulingIgnoredDuringExecution: nil,
-		},
-	}
-
-	cluster := v1beta1.KafkaClusterSpec{
-		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
-			"default": {
-				Affinity: &corev1.Affinity{
-					PodAntiAffinity: &corev1.PodAntiAffinity{
-						RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-							{
-								LabelSelector: &metav1.LabelSelector{
-									MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
-								},
-								TopologyKey: "kubernetes.io/hostname",
-							},
-						},
-					},
-					NodeAffinity: &corev1.NodeAffinity{
-						RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
-							NodeSelectorTerms: []corev1.NodeSelectorTerm{
-								{
-									MatchExpressions: nil,
-									MatchFields: []corev1.NodeSelectorRequirement{
-										{
-											Key:      "fruit",
-											Operator: "in",
-											Values:   []string{"apple"},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	broker := v1beta1.Broker{
-		Id:                0,
-		BrokerConfigGroup: "default",
-	}
-
-	config, err := GetBrokerConfig(broker, cluster)
-	if err != nil {
-		t.Error(err)
-	}
-
-	assert.DeepEqual(t, expected, config.Affinity)
-}
-
-// the equal values got duplicated
-func TestGetBrokerConfigAffinityMergeEqualPodAntiAffinity(t *testing.T) {
-	expected := &corev1.Affinity{
-		NodeAffinity: nil,
-		PodAffinity:  nil,
-		PodAntiAffinity: &corev1.PodAntiAffinity{
-			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-				{
-					LabelSelector: &metav1.LabelSelector{
-						MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
-					},
-					TopologyKey: "kubernetes.io/hostname",
-				},
-			},
-			PreferredDuringSchedulingIgnoredDuringExecution: nil,
-		},
-	}
-
-	broker := v1beta1.Broker{
-		Id:                0,
-		BrokerConfigGroup: "default",
-		BrokerConfig: &v1beta1.BrokerConfig{
-			Affinity: &corev1.Affinity{
-				PodAntiAffinity: &corev1.PodAntiAffinity{
-					RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-						{
-							LabelSelector: &metav1.LabelSelector{
-								MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_broker"},
-							},
-							TopologyKey: "kubernetes.io/hostname",
-						},
-					},
-				},
-			},
-		},
-	}
-
-	cluster := v1beta1.KafkaClusterSpec{
-		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
-			"default": {
-				Affinity: &corev1.Affinity{
-					PodAntiAffinity: &corev1.PodAntiAffinity{
-						RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
-							{
-								LabelSelector: &metav1.LabelSelector{
-									MatchLabels: map[string]string{"app": "kafka", "kafka_cr": "kafka_config_group"},
-								},
-								TopologyKey: "kubernetes.io/hostname",
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	config, err := GetBrokerConfig(broker, cluster)
-	if err != nil {
-		t.Error(err)
-	}
-	assert.DeepEqual(t, expected, config.Affinity)
-}
-
-func TestGetBrokerConfig(t *testing.T) {
-	expected := &v1beta1.BrokerConfig{
-		StorageConfigs: []v1beta1.StorageConfig{
-			{
-				MountPath: "kafka-test/log",
-				PvcSpec: &corev1.PersistentVolumeClaimSpec{
-					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-					Resources: corev1.ResourceRequirements{
-						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-					},
-				},
-			},
-			{
-				MountPath: "kafka-test1/log",
-				PvcSpec: &corev1.PersistentVolumeClaimSpec{
-					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-					Resources: corev1.ResourceRequirements{
-						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-					},
-				},
-			},
-		},
-	}
-	broker := v1beta1.Broker{
-		Id:                0,
-		BrokerConfigGroup: "default",
-		BrokerConfig: &v1beta1.BrokerConfig{
-			StorageConfigs: []v1beta1.StorageConfig{
-				{
-					MountPath: "kafka-test/log",
-					PvcSpec: &corev1.PersistentVolumeClaimSpec{
-						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-						Resources: corev1.ResourceRequirements{
-							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	cluster := v1beta1.KafkaClusterSpec{
-		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
-			"default": {
-				StorageConfigs: []v1beta1.StorageConfig{
-					{
-						MountPath: "kafka-test1/log",
-						PvcSpec: &corev1.PersistentVolumeClaimSpec{
-							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-							Resources: corev1.ResourceRequirements{
-								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	result, err := GetBrokerConfig(broker, cluster)
-	if err != nil {
-		t.Error("Error GetBrokerConfig throw an unexpected error")
-	}
-	if !reflect.DeepEqual(result, expected) {
-		t.Error("Expected:", expected, "Got:", result)
-	}
-}
-
-func TestGetBrokerConfigUniqueStorage(t *testing.T) {
-	expected := &v1beta1.BrokerConfig{
-		StorageConfigs: []v1beta1.StorageConfig{
-			{
-				MountPath: "kafka-test/log",
-				PvcSpec: &corev1.PersistentVolumeClaimSpec{
-					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-					Resources: corev1.ResourceRequirements{
-						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("20Gi")},
-					},
-				},
-			},
-			{
-				MountPath: "kafka-test1/log",
-				PvcSpec: &corev1.PersistentVolumeClaimSpec{
-					AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-					Resources: corev1.ResourceRequirements{
-						Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-					},
-				},
-			},
-		},
-	}
-	broker := v1beta1.Broker{
-		Id:                0,
-		BrokerConfigGroup: "default",
-		BrokerConfig: &v1beta1.BrokerConfig{
-			StorageConfigs: []v1beta1.StorageConfig{
-				{
-					MountPath: "kafka-test/log",
-					PvcSpec: &corev1.PersistentVolumeClaimSpec{
-						AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-						Resources: corev1.ResourceRequirements{
-							Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("20Gi")},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	cluster := v1beta1.KafkaClusterSpec{
-		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
-			"default": {
-				StorageConfigs: []v1beta1.StorageConfig{
-					{
-						MountPath: "kafka-test1/log",
-						PvcSpec: &corev1.PersistentVolumeClaimSpec{
-							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-							Resources: corev1.ResourceRequirements{
-								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-							},
-						},
-					},
-					{
-						MountPath: "kafka-test/log",
-						PvcSpec: &corev1.PersistentVolumeClaimSpec{
-							AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
-							Resources: corev1.ResourceRequirements{
-								Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("10Gi")},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-
-	result, err := GetBrokerConfig(broker, cluster)
-	if err != nil {
-		t.Error("Error GetBrokerConfig throw an unexpected error")
-	}
-	if !reflect.DeepEqual(result, expected) {
-		t.Error("Expected:", expected, "Got:", result)
-	}
-}
-
 func TestIntstrPointer(t *testing.T) {
 	i := int(10)
 	ptr := IntstrPointer(i)