Repository: banzaicloud/kafka-operator
Pull request #547 (2021-02-02 08:56:53)

refactor: get bootstrap brokers
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | no
| API breaks?     | no
| Deprecations?   | no
| License         | Apache 2.0

### What's in this PR?
Refactor the way how the list of Kafka bootstrap servers is compiled and retrieved.

### 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)


chrisgacsal: > Wouldn't be more efficient if `cruise.control.metrics.reporter.bootstrap.servers` is pointed at the local broker (the broker the reporter runs in) instead of `kafka-all-brokers`?

I do not think it would make much difference as the `bootstrap.servers` is used by the Kafka client only for discovery purposes. I have no preference though, so either way is fine by me.

> Also, should the `cruise.control.metrics.reporter` run in each broker or only in the controller broker?

So each broker reports their own metrics only which means that we need `cruise.control.metrics.reporter` to be invoked as part of all the brokers in the cluster in order to make metrics for the whole cluster available for Cruise Control.

@stoader Thanks for the review. Please find my answers inline.

Patch:
diff --git a/.gitignore b/.gitignore
index 1e9fade3..a12d00b8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -38,3 +38,5 @@ bin/*
 cover.out
 
 manager_image_patch.yaml
+
+.idea
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index a97be708..f781fdac 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -43,11 +43,10 @@ func (r *Reconciler) configMap(clientPass, capacityConfig string, log logr.Logge
 	}
 	ccConfig.Merge(conf)
 
-	// Set bootstrap servers
-	bootstrapServers := fmt.Sprintf("%s:%d",
-		generateBootstrapServer(r.KafkaCluster.Spec.HeadlessServiceEnabled, r.KafkaCluster.Name),
-		generateBootstrapServerPort(log, r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
-	)
+	bootstrapServers, err := kafkautils.GetBootstrapServersService(r.KafkaCluster)
+	if err != nil {
+		log.Error(err, "getting Kafka bootstrap servers for Cruise Control failed")
+	}
 	if err = ccConfig.Set("bootstrap.servers", bootstrapServers); err != nil {
 		log.Error(err, "setting bootstrap.servers in Cruise Control configuration failed", "config", bootstrapServers)
 	}
@@ -105,24 +104,6 @@ func generateSSLConfig(l *v1beta1.ListenersConfig, clientPass string, log logr.L
 	return sslConf
 }
 
-func generateBootstrapServer(headlessEnabled bool, clusterName string) string {
-	if headlessEnabled {
-		return fmt.Sprintf(kafkautils.HeadlessServiceTemplate, clusterName)
-	}
-	return fmt.Sprintf(kafkautils.AllBrokerServiceTemplate, clusterName)
-}
-
-func generateBootstrapServerPort(log logr.Logger, listenerConfig []v1beta1.InternalListenerConfig) int32 {
-	for _, listener := range listenerConfig {
-		if listener.UsedForInnerBrokerCommunication {
-			return listener.ContainerPort
-		}
-	}
-	// This should never happen since there should be least one listener for inner broker communication
-	log.V(1).Info("listener for inner communication not found, falling back to first listener")
-	return listenerConfig[0].ContainerPort
-}
-
 const (
 	storageConfigCPUDefaultValue   = "100"
 	storageConfigNWINDefaultValue  = "125000"
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 3c42c1f5..5fd332a7 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -27,7 +27,7 @@ import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
-	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
+	kafkautils "github.com/banzaicloud/kafka-operator/pkg/util/kafka"
 	zookeeperutils "github.com/banzaicloud/kafka-operator/pkg/util/zookeeper"
 	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 )
@@ -99,7 +99,10 @@ func (r *Reconciler) getConfigProperties(bConfig *v1beta1.BrokerConfig, id int32
 	if err := config.Set("metric.reporters", "com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"); err != nil {
 		log.Error(err, "setting metric.reporters in broker configuration resulted an error")
 	}
-	bootstrapServers := getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled)
+	bootstrapServers, err := kafkautils.GetBootstrapServersService(r.KafkaCluster)
+	if err != nil {
+		log.Error(err, "getting Kafka bootstrap servers for Cruise Control failed")
+	}
 	if err := config.Set("cruise.control.metrics.reporter.bootstrap.servers", bootstrapServers); err != nil {
 		log.Error(err, "setting cruise.control.metrics.reporter.bootstrap.servers in broker configuration resulted an error")
 	}
@@ -143,12 +146,12 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, ext
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(brokerConfigTemplate+"-%d", r.KafkaCluster.Name, id),
 			util.MergeLabels(
-				kafka.LabelsForKafka(r.KafkaCluster.Name),
+				kafkautils.LabelsForKafka(r.KafkaCluster.Name),
 				map[string]string{"brokerId": fmt.Sprintf("%d", id)},
 			),
 			r.KafkaCluster,
 		),
-		Data: map[string]string{kafka.ConfigPropertyName: r.generateBrokerConfig(id, brokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)},
+		Data: map[string]string{kafkautils.ConfigPropertyName: r.generateBrokerConfig(id, brokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)},
 	}
 	if brokerConfig.Log4jConfig != "" {
 		brokerConf.Data["log4j.properties"] = brokerConfig.Log4jConfig
@@ -240,23 +243,6 @@ func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger)
 	return config
 }
 
-func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32, domain, namespace, crName string, headlessServiceEnabled bool) string {
-
-	internalListener := ""
-
-	for _, iListener := range iListeners {
-		if iListener.UsedForInnerBrokerCommunication {
-			if headlessServiceEnabled {
-				internalListener = fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, domain, iListener.ContainerPort)
-			} else {
-				internalListener = fmt.Sprintf("%s://%s-%d.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, namespace, domain, iListener.ContainerPort)
-			}
-		}
-	}
-
-	return internalListener
-}
-
 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 {
 	// Parse cluster-wide readonly configuration
 	finalBrokerConfig, err := properties.NewFromString(r.KafkaCluster.Spec.ReadOnlyConfig)
diff --git a/pkg/util/kafka/common.go b/pkg/util/kafka/common.go
index fce2a05b..64141bde 100644
--- a/pkg/util/kafka/common.go
+++ b/pkg/util/kafka/common.go
@@ -18,6 +18,8 @@ import (
 	"fmt"
 	"strings"
 
+	"emperror.dev/errors"
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
 	"github.com/go-logr/logr"
 
@@ -166,3 +168,68 @@ func newListenerSecurityProtocolMap(s string) listenerSecurityProtocolMap {
 
 	return listenerSecProtoMap
 }
+
+const (
+	// BrokerHostnameTemplate defines the hostname template for Kafka brokers in the following format:
+	// 	<KAFKA_CLUSTER_NAME>-<BROKER_ID>
+	BrokerHostnameTemplate = "%s-%d"
+	// BrokerHostnameTemplate defines the domain template for Kafka brokers in the following format:
+	// 	<K8S_NAMESPACE>.svc.<K8S_CLUSTER_DOMAIN>
+	ServiceDomainNameTemplate = "%s.svc.%s"
+)
+
+func GetClusterServiceDomainName(cluster *v1beta1.KafkaCluster) string {
+	return fmt.Sprintf(ServiceDomainNameTemplate, cluster.Namespace, cluster.Spec.GetKubernetesClusterDomain())
+}
+
+func GetBrokerServiceFqdn(cluster *v1beta1.KafkaCluster, broker *v1beta1.Broker) string {
+	hostname := fmt.Sprintf(BrokerHostnameTemplate, cluster.Name, broker.Id)
+	svcDomainName := GetClusterServiceDomainName(cluster)
+	return fmt.Sprintf("%s.%s", hostname, svcDomainName)
+}
+
+func GetClusterServiceFqdn(cluster *v1beta1.KafkaCluster) string {
+	tmpl := AllBrokerServiceTemplate
+	if cluster.Spec.HeadlessServiceEnabled {
+		tmpl = HeadlessServiceTemplate
+	}
+	return fmt.Sprintf("%s.%s",
+		fmt.Sprintf(tmpl, cluster.Name),
+		GetClusterServiceDomainName(cluster))
+}
+
+func GetBootstrapServers(cluster *v1beta1.KafkaCluster) (string, error) {
+	return getBootstrapServers(cluster, false)
+}
+
+func GetBootstrapServersService(cluster *v1beta1.KafkaCluster) (string, error) {
+	return getBootstrapServers(cluster, true)
+}
+
+func getBootstrapServers(cluster *v1beta1.KafkaCluster, useService bool) (string, error) {
+	var listener v1beta1.InternalListenerConfig
+	var bootstrapServersList []string
+
+	for _, lc := range cluster.Spec.ListenersConfig.InternalListeners {
+		if lc.UsedForInnerBrokerCommunication && !lc.UsedForControllerCommunication {
+			listener = lc
+			break
+		}
+	}
+
+	if listener.Name == "" {
+		return "", errors.New("no suitable listener found for using as Kafka bootstrap server configuration")
+	}
+
+	if useService {
+		bootstrapServersList = append(bootstrapServersList,
+			fmt.Sprintf("%s:%d", GetClusterServiceFqdn(cluster), listener.ContainerPort))
+	} else {
+		for _, broker := range cluster.Spec.Brokers {
+			fqdn := GetBrokerServiceFqdn(cluster, &broker)
+			bootstrapServersList = append(bootstrapServersList,
+				fmt.Sprintf("%s:%d", fqdn, listener.ContainerPort))
+		}
+	}
+	return strings.Join(bootstrapServersList, ","), nil
+}


Test patch:
diff --git a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
index 45b5bdc1..39bf9168 100644
--- a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
+++ b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
@@ -110,10 +110,10 @@ func expectCruiseControlConfigMap(kafkaCluster *v1beta1.KafkaCluster) {
 	Expect(configMap.Labels).To(HaveKeyWithValue("app", "cruisecontrol"))
 	Expect(configMap.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
 
-	Expect(configMap.Data).To(HaveKeyWithValue("cruisecontrol.properties", fmt.Sprintf(`bootstrap.servers=%s-all-broker:29092
+	Expect(configMap.Data).To(HaveKeyWithValue("cruisecontrol.properties", fmt.Sprintf(`bootstrap.servers=%s-all-broker.%s.%s:29092
 some.config=value
 zookeeper.connect=/
-`, kafkaCluster.Name)))
+`, kafkaCluster.Name, kafkaCluster.Namespace, "svc.cluster.local")))
 	Expect(configMap.Data).To(HaveKeyWithValue("capacity.json", `{
     "brokerCapacities": [
         {
diff --git a/controllers/tests/kafkacluster_controller_kafka_test.go b/controllers/tests/kafkacluster_controller_kafka_test.go
index 60a6311a..2e5f02ce 100644
--- a/controllers/tests/kafkacluster_controller_kafka_test.go
+++ b/controllers/tests/kafkacluster_controller_kafka_test.go
@@ -165,7 +165,7 @@ func expectKafkaBrokerConfigmap(kafkaCluster *v1beta1.KafkaCluster, broker v1bet
 	Expect(configMap.Data).To(HaveKeyWithValue("broker-config", fmt.Sprintf(`advertised.listeners=TEST://test.host.com:%d,INTERNAL://kafkacluster-1-%d.kafka-1.svc.cluster.local:29092,CONTROLLER://kafkacluster-1-%d.kafka-1.svc.cluster.local:29093
 broker.id=%d
 control.plane.listener.name=CONTROLLER
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafkacluster-1-%d.kafka-1.svc.cluster.local:29092
+cruise.control.metrics.reporter.bootstrap.servers=kafkacluster-1-all-broker.kafka-1.svc.cluster.local:29092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT,TEST:
@@ -173,7 +173,7 @@ listeners=INTERNAL://:29092,CONTROLLER://:29093,TEST://:9094
 log.dirs=/kafka-logs/kafka
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
 zookeeper.connect=/
-`, 19090+broker.Id, broker.Id, broker.Id, broker.Id, broker.Id)))
+`, 19090+broker.Id, broker.Id, broker.Id, broker.Id)))
 
 	// assert log4j?
 }
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 2fb41186..78b0b5ed 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -52,7 +52,7 @@ func TestGenerateBrokerConfig(t *testing.T) {
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -72,7 +72,7 @@ zookeeper.connect=example.zk:2181/`,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -92,7 +92,7 @@ zookeeper.connect=example.zk:2181/kafka`,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -112,7 +112,7 @@ zookeeper.connect=example.zk:2181/`,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -137,7 +137,7 @@ zookeeper.connect=example.zk:2181,example.zk-1:2181/kafka`,
 			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -158,7 +158,7 @@ zookeeper.connect=example.zk:2181/`,
 			advertisedListenerAddress: `kafka-0.kafka.svc.foo.bar:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.foo.bar:9092
 broker.id=0
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.foo.bar:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.foo.bar:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
@@ -188,7 +188,7 @@ auto.create.topics.enable=true
 auto.create.topics.enable=true
 broker.id=0
 control.plane.listener.name=thisisatest
-cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
+cruise.control.metrics.reporter.bootstrap.servers=kafka-all-broker.kafka.svc.cluster.local:9092
 cruise.control.metrics.reporter.kubernetes.mode=true
 inter.broker.listener.name=INTERNAL
 listener.security.protocol.map=INTERNAL:PLAINTEXT
diff --git a/pkg/util/kafka/common_test.go b/pkg/util/kafka/common_test.go
index 3f509b2b..ee973e92 100644
--- a/pkg/util/kafka/common_test.go
+++ b/pkg/util/kafka/common_test.go
@@ -17,8 +17,12 @@ package kafka
 import (
 	"testing"
 
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	properties "github.com/banzaicloud/kafka-operator/properties/pkg"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 func TestShouldRefreshOnlyPerBrokerConfigs(t *testing.T) {
@@ -102,3 +106,243 @@ ssl.client.auth=modified_value_4
 		}
 	}
 }
+
+const defaultBrokerConfigGroup = "default"
+
+var MinimalKafkaCluster = &v1beta1.KafkaCluster{
+	ObjectMeta: metav1.ObjectMeta{
+		Name:      "kafka-cluster",
+		Namespace: "kafka-ns",
+	},
+	Spec: v1beta1.KafkaClusterSpec{
+		ListenersConfig: v1beta1.ListenersConfig{
+			ExternalListeners: []v1beta1.ExternalListenerConfig{
+				{
+					CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Name:          "test",
+						ContainerPort: 9094,
+					},
+					ExternalStartingPort: 19090,
+					AccessMethod:         corev1.ServiceTypeLoadBalancer,
+				},
+			},
+			InternalListeners: []v1beta1.InternalListenerConfig{
+				{
+					CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Type:          "plaintext",
+						Name:          "internal",
+						ContainerPort: 29092,
+					},
+					UsedForInnerBrokerCommunication: true,
+				},
+				{
+					CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Type:          "plaintext",
+						Name:          "controller",
+						ContainerPort: 29093,
+					},
+					UsedForInnerBrokerCommunication: false,
+					UsedForControllerCommunication:  true,
+				},
+			},
+		},
+		BrokerConfigGroups: map[string]v1beta1.BrokerConfig{
+			defaultBrokerConfigGroup: {
+				StorageConfigs: []v1beta1.StorageConfig{
+					{
+						MountPath: "/kafka-logs",
+						PvcSpec: &corev1.PersistentVolumeClaimSpec{
+							AccessModes: []corev1.PersistentVolumeAccessMode{
+								corev1.ReadWriteOnce,
+							},
+							Resources: corev1.ResourceRequirements{
+								Requests: map[corev1.ResourceName]resource.Quantity{
+									corev1.ResourceStorage: resource.MustParse("10Gi"),
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+		Brokers: []v1beta1.Broker{
+			{
+				Id:                0,
+				BrokerConfigGroup: defaultBrokerConfigGroup,
+			},
+			{
+				Id:                1,
+				BrokerConfigGroup: defaultBrokerConfigGroup,
+			},
+			{
+				Id:                2,
+				BrokerConfigGroup: defaultBrokerConfigGroup,
+			},
+		},
+	},
+}
+
+func TestGetClusterServiceDomainName(t *testing.T) {
+	t.Run("Without cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Namespace = "kafka-ns"
+		cluster.Spec.KubernetesClusterDomain = ""
+		svcName := GetClusterServiceDomainName(cluster)
+		expected := "kafka-ns.svc.cluster.local"
+
+		if svcName != expected {
+			t.Errorf("Mismatch in service domain name. Expected: %v, got %v", expected, svcName)
+		}
+	})
+
+	t.Run("With cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Namespace = "kafka-ns2"
+		cluster.Spec.KubernetesClusterDomain = "foo.bar"
+		svcName := GetClusterServiceDomainName(cluster)
+		expected := "kafka-ns2.svc.foo.bar"
+
+		if svcName != expected {
+			t.Errorf("Mismatch in service domain name. Expected: %v, got %v", expected, svcName)
+		}
+	})
+}
+
+func TestGetBrokerServiceFqdn(t *testing.T) {
+	broker := &v1beta1.Broker{
+		Id:                1,
+		BrokerConfigGroup: defaultBrokerConfigGroup,
+	}
+
+	t.Run("Without cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Name = "kafka-cluster"
+		cluster.Namespace = "kafka-ns"
+		cluster.Spec.KubernetesClusterDomain = ""
+		fqdn := GetBrokerServiceFqdn(cluster, broker)
+		expectedFqdn := "kafka-cluster-1.kafka-ns.svc.cluster.local"
+
+		if fqdn != expectedFqdn {
+			t.Errorf("Mismatch in broker service fqdn. Expected: %v, got %v", expectedFqdn, fqdn)
+		}
+	})
+
+	t.Run("With cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Name = "kafka-cluster1"
+		cluster.Namespace = "kafka-ns2"
+		cluster.Spec.KubernetesClusterDomain = "foo.bar"
+		fqdn := GetBrokerServiceFqdn(cluster, broker)
+		expectedFqdn := "kafka-cluster1-1.kafka-ns2.svc.foo.bar"
+
+		if fqdn != expectedFqdn {
+			t.Errorf("Mismatch in broker service fqdn. Expected: %v, got %v", expectedFqdn, fqdn)
+		}
+	})
+}
+
+func TestGetBootstrapServersService(t *testing.T) {
+
+	t.Run("Without headless service and cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.HeadlessServiceEnabled = false
+		cluster.Spec.KubernetesClusterDomain = ""
+		expected := "kafka-cluster-all-broker.kafka-ns.svc.cluster.local:29092"
+
+		bootstrapServers, err := GetBootstrapServersService(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in cluster service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+
+	t.Run("With headless service and cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.HeadlessServiceEnabled = true
+		cluster.Spec.KubernetesClusterDomain = "foo.bar"
+		expected := "kafka-cluster-headless.kafka-ns.svc.foo.bar:29092"
+
+		bootstrapServers, err := GetBootstrapServersService(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in cluster service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+
+	t.Run("With headless service but without cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.HeadlessServiceEnabled = true
+		cluster.Spec.KubernetesClusterDomain = ""
+		expected := "kafka-cluster-headless.kafka-ns.svc.cluster.local:29092"
+
+		bootstrapServers, err := GetBootstrapServersService(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in cluster service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+
+	t.Run("With cluster domain override but without headless service", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.HeadlessServiceEnabled = false
+		cluster.Spec.KubernetesClusterDomain = "foo.bar"
+		expected := "kafka-cluster-all-broker.kafka-ns.svc.foo.bar:29092"
+
+		bootstrapServers, err := GetBootstrapServersService(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in cluster service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+}
+
+func TestGetBootstrapServers(t *testing.T) {
+
+	t.Run("Without cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.KubernetesClusterDomain = ""
+		expected := "kafka-cluster-0.kafka-ns.svc.cluster.local:29092,kafka-cluster-1.kafka-ns.svc.cluster.local:29092,kafka-cluster-2.kafka-ns.svc.cluster.local:29092"
+
+		bootstrapServers, err := GetBootstrapServers(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in broker service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+
+	t.Run("With cluster domain override", func(t *testing.T) {
+		cluster := MinimalKafkaCluster.DeepCopy()
+		cluster.Spec.KubernetesClusterDomain = "foo.bar"
+		expected := "kafka-cluster-0.kafka-ns.svc.foo.bar:29092,kafka-cluster-1.kafka-ns.svc.foo.bar:29092,kafka-cluster-2.kafka-ns.svc.foo.bar:29092"
+
+		bootstrapServers, err := GetBootstrapServers(cluster)
+
+		if err != nil {
+			t.Errorf("Should not return error. Got %v", err)
+		}
+
+		if bootstrapServers != expected {
+			t.Errorf("Mismatch in cluster service bootstrap servers string. Expected: %v, got %v", expected, bootstrapServers)
+		}
+	})
+}