Repository: banzaicloud/kafka-operator
Pull request #601 (2021-05-03 16:08:42)

Add kafka broker version to status field and update jmx exporter version
| Q               | A
| --------------- | ---
| Bug fix?        | no|
| New feature?    |yes|
| API breaks?     | no|
| Deprecations?   | no|
| 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. -->
This PR adds broker version to the kafka cluster cr.
It also updates the jmx exporter to version 0.15.0 instead of 0.14.0.
And slightly modifies the jmx exporter config to be able to parse the version properly.


### 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
- [X] User guide and development docs updated (if needed)



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/jmxextractor/extractor.go b/pkg/jmxextractor/extractor.go
new file mode 100644
index 00000000..5313abe1
--- /dev/null
+++ b/pkg/jmxextractor/extractor.go
@@ -0,0 +1,112 @@
+// 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 jmxextractor
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"regexp"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
+	"github.com/banzaicloud/kafka-operator/pkg/util"
+	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
+
+	"github.com/go-logr/logr"
+)
+
+const (
+	headlessServiceJMXTemplate = "http://%s-%d." + kafka.HeadlessServiceTemplate + ".%s.svc.%s:%d"
+	serviceJMXTemplate         = "http://%s-%d.%s.svc.%s:%d"
+	versionRegexGroup          = "version"
+)
+
+var newJMXExtractor = createNewJMXExtractor
+var jmxMetricRegex = regexp.MustCompile(
+	fmt.Sprintf(`kafka_server_app_info_version{broker_id=\"[0-9]+\",version=\"(?P<%s>[0-9]+.[0-9]+.[0-9]+)\"`, versionRegexGroup))
+
+type JMXExtractor interface {
+	ExtractDockerImageAndVersion(brokerId int32, brokerConfig *v1beta1.BrokerConfig,
+		clusterImage string, headlessServiceEnabled bool) (*v1beta1.KafkaVersion, error)
+}
+
+type jmxExtractor struct {
+	JMXExtractor
+	clusterNamespace        string
+	kubernetesClusterDomain string
+	clusterName             string
+	log                     logr.Logger
+}
+
+func createNewJMXExtractor(namespace, kubernetesClusterDomain, clusterName string, log logr.Logger) JMXExtractor {
+	return &jmxExtractor{
+		clusterNamespace:        namespace,
+		kubernetesClusterDomain: kubernetesClusterDomain,
+		clusterName:             clusterName,
+		log:                     log,
+	}
+}
+func createMockJMXExtractor(namespace, kubernetesClusterDomain, clusterName string, log logr.Logger) JMXExtractor {
+	return &mockJmxExtractor{}
+}
+
+func NewJMXExtractor(namespace, kubernetesClusterDomain, clusterName string, log logr.Logger) JMXExtractor {
+	return newJMXExtractor(namespace, kubernetesClusterDomain, clusterName, log)
+}
+
+func NewMockJMXExtractor() {
+	newJMXExtractor = createMockJMXExtractor
+}
+
+func (exp *jmxExtractor) ExtractDockerImageAndVersion(brokerId int32, brokerConfig *v1beta1.BrokerConfig,
+	clusterImage string, headlessServiceEnabled bool) (*v1beta1.KafkaVersion, error) {
+	var requestURL string
+	if headlessServiceEnabled {
+		requestURL =
+			fmt.Sprintf(headlessServiceJMXTemplate,
+				exp.clusterName, brokerId, exp.clusterName, exp.clusterNamespace,
+				exp.kubernetesClusterDomain, 9020)
+	} else {
+		requestURL = fmt.Sprintf(serviceJMXTemplate, exp.clusterName, brokerId,
+			exp.clusterNamespace, exp.kubernetesClusterDomain, 9020)
+	}
+	rsp, err := http.Get(requestURL)
+	if err != nil {
+		exp.log.Error(err, fmt.Sprintf("error during talking to broker-%d", brokerId))
+		return nil, errorfactory.New(errorfactory.BrokersNotReady{}, err, "unable to talk to ...")
+	}
+	defer func() {
+		closeErr := rsp.Body.Close()
+		if closeErr != nil {
+			exp.log.Error(closeErr, "could not close client")
+		}
+	}()
+
+	body, err := ioutil.ReadAll(rsp.Body)
+	if err != nil {
+		return nil, err
+	}
+	index := jmxMetricRegex.SubexpIndex(versionRegexGroup)
+	var version string
+	if index > -1 {
+		if metrics := jmxMetricRegex.FindStringSubmatch(string(body)); len(metrics) > index {
+			version = metrics[index]
+		}
+	}
+
+	brokerImage := util.GetBrokerImage(brokerConfig, clusterImage)
+	return &v1beta1.KafkaVersion{Version: version, Image: brokerImage}, nil
+}
diff --git a/pkg/jmxextractor/mock_extractor.go b/pkg/jmxextractor/mock_extractor.go
new file mode 100644
index 00000000..893f14c2
--- /dev/null
+++ b/pkg/jmxextractor/mock_extractor.go
@@ -0,0 +1,24 @@
+// 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 jmxextractor
+
+import "github.com/banzaicloud/kafka-operator/api/v1beta1"
+
+type mockJmxExtractor struct{}
+
+func (exp *mockJmxExtractor) ExtractDockerImageAndVersion(brokerId int32, brokerConfig *v1beta1.BrokerConfig,
+	clusterImage string, headlessServiceEnabled bool) (*v1beta1.KafkaVersion, error) {
+	return &v1beta1.KafkaVersion{Image: clusterImage, Version: "2.7.0"}, nil
+}
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..3754fecf 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -37,6 +37,7 @@ import (
 	"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/jmxextractor"
 	"github.com/banzaicloud/kafka-operator/pkg/k8sutil"
 	"github.com/banzaicloud/kafka-operator/pkg/kafkaclient"
 	"github.com/banzaicloud/kafka-operator/pkg/pki"
@@ -206,7 +207,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 +230,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")
 		}
@@ -565,10 +566,18 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod,
 	return nil
 }
 
-func (r *Reconciler) updateStatusWithDockerImageAndVersion(brokerId int32, brokerConfig *v1beta1.BrokerConfig, log logr.Logger) error {
-	brokerImage := util.GetBrokerImage(brokerConfig, r.KafkaCluster.Spec.GetClusterImage())
-	err := k8sutil.UpdateBrokerStatus(r.Client, []string{strconv.Itoa(int(brokerId))}, r.KafkaCluster,
-		v1beta1.KafkaVersion{Image: brokerImage}, log)
+func (r *Reconciler) updateStatusWithDockerImageAndVersion(brokerId int32, brokerConfig *v1beta1.BrokerConfig,
+	log logr.Logger) error {
+	jmxExp := jmxextractor.NewJMXExtractor(r.KafkaCluster.GetNamespace(),
+		r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.GetName(), log)
+
+	kafkaVersion, err := jmxExp.ExtractDockerImageAndVersion(brokerId, brokerConfig,
+		r.KafkaCluster.Spec.GetClusterImage(), r.KafkaCluster.Spec.HeadlessServiceEnabled)
+	if err != nil {
+		return err
+	}
+	err = k8sutil.UpdateBrokerStatus(r.Client, []string{strconv.Itoa(int(brokerId))}, r.KafkaCluster,
+		*kafkaVersion, log)
 	if err != nil {
 		return err
 	}
@@ -879,7 +888,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 +911,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/scale/mock_scaler.go b/pkg/scale/mock_scaler.go
index b59fc8a4..cac6b500 100644
--- a/pkg/scale/mock_scaler.go
+++ b/pkg/scale/mock_scaler.go
@@ -18,13 +18,7 @@ import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
 
-type mockCruiseControlScaler struct {
-	CruiseControlScaler
-	//namespace               string
-	//kubernetesClusterDomain string
-	//endpoint                string
-	//clusterName             string
-}
+type mockCruiseControlScaler struct{}
 
 func (mc *mockCruiseControlScaler) GetLiveKafkaBrokersFromCruiseControl(brokerIds []string) ([]string, error) {
 	return nil, nil
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..3af62614 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"
@@ -811,7 +814,7 @@ func (mConfig *MonitoringConfig) GetImage() string {
 	if mConfig.JmxImage != "" {
 		return mConfig.JmxImage
 	}
-	return "ghcr.io/banzaicloud/jmx-javaagent:0.14.0"
+	return "ghcr.io/banzaicloud/jmx-javaagent:0.15.0"
 }
 
 // GetPathToJar returns the path in the used Image for Prometheus JMX exporter
@@ -819,7 +822,7 @@ func (mConfig *MonitoringConfig) GetPathToJar() string {
 	if mConfig.PathToJar != "" {
 		return mConfig.PathToJar
 	}
-	return "/opt/jmx_exporter/jmx_prometheus_javaagent-0.14.0.jar"
+	return "/jmx_prometheus_javaagent.jar"
 }
 
 // GetKafkaJMXExporterConfig returns the config for Kafka Prometheus JMX exporter
@@ -831,11 +834,13 @@ func (mConfig *MonitoringConfig) GetKafkaJMXExporterConfig() string {
 	return `lowercaseOutputName: true
 rules:
 # Special cases and very specific rules
-- pattern : 'kafka.server<type=app-info><>version: (.*)'
-  name: kafka_server_app_info
-  value: 1.0
+- pattern: 'kafka.server<type=(app-info), id=(\d+)><>(Version): ([-.~+\w\d]+)'
+  name: kafka_server_$1_$3
+  type: COUNTER
   labels:
-    version: "$1"
+    broker_id: $2
+    version: $4
+  value: 1.0
 - pattern : kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
   name: kafka_server_$1_$2
   type: GAUGE
@@ -1055,6 +1060,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/controllers/tests/kafkacluster_controller_cruisecontrol_test.go b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
index 6b49562e..2b85262b 100644
--- a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
+++ b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
@@ -256,8 +256,8 @@ func expectCruiseControlDeployment(kafkaCluster *v1beta1.KafkaCluster) {
 	Expect(deployment.Spec.Template.Spec.InitContainers).To(HaveLen(1))
 	initContainer := deployment.Spec.Template.Spec.InitContainers[0]
 	Expect(initContainer.Name).To(Equal("jmx-exporter"))
-	Expect(initContainer.Image).To(Equal("ghcr.io/banzaicloud/jmx-javaagent:0.14.0"))
-	Expect(initContainer.Command).To(Equal([]string{"cp", "/opt/jmx_exporter/jmx_prometheus_javaagent-0.14.0.jar", "/opt/jmx-exporter/jmx_prometheus.jar"}))
+	Expect(initContainer.Image).To(Equal("ghcr.io/banzaicloud/jmx-javaagent:0.15.0"))
+	Expect(initContainer.Command).To(Equal([]string{"cp", "/jmx_prometheus_javaagent.jar", "/opt/jmx-exporter/jmx_prometheus.jar"}))
 	Expect(initContainer.VolumeMounts).To(ConsistOf(corev1.VolumeMount{
 		Name:      "jmx-jar-data",
 		MountPath: "/opt/jmx-exporter/",
diff --git a/controllers/tests/kafkacluster_controller_kafka_test.go b/controllers/tests/kafkacluster_controller_kafka_test.go
index 2c9b91bc..1be9abbd 100644
--- a/controllers/tests/kafkacluster_controller_kafka_test.go
+++ b/controllers/tests/kafkacluster_controller_kafka_test.go
@@ -434,4 +434,8 @@ func expectKafkaCRStatus(kafkaCluster *v1beta1.KafkaCluster) {
 			},
 		},
 	}))
+	for _, brokerState := range kafkaCluster.Status.BrokersState {
+		Expect(brokerState.Version).To(Equal("2.7.0"))
+		Expect(brokerState.Image).To(Equal(kafkaCluster.Spec.GetClusterImage()))
+	}
 }
diff --git a/controllers/tests/suite_test.go b/controllers/tests/suite_test.go
index c39bab37..08d84b35 100644
--- a/controllers/tests/suite_test.go
+++ b/controllers/tests/suite_test.go
@@ -62,6 +62,7 @@ import (
 	banzaicloudv1alpha1 "github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	banzaicloudv1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/controllers"
+	"github.com/banzaicloud/kafka-operator/pkg/jmxextractor"
 	"github.com/banzaicloud/kafka-operator/pkg/kafkaclient"
 	"github.com/banzaicloud/kafka-operator/pkg/scale"
 	// +kubebuilder:scaffold:imports
@@ -134,6 +135,7 @@ var _ = BeforeSuite(func(done Done) {
 	Expect(mgr).ToNot(BeNil())
 
 	scale.MockNewCruiseControlScaler()
+	jmxextractor.NewMockJMXExtractor()
 
 	mockKafkaClients = make(map[types.NamespacedName]kafkaclient.KafkaClient)
 
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)