Repository: banzaicloud/kafka-operator
Pull request #519 (2020-12-10 16:13:43)

Introduce listener status field
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | yes
| 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. -->
Introducing listener status field which contains listener addresses. Client application can use this to connect to the cluster using either an internal or and external listener.

### Why?
<!-- Which problem does the PR fix? (Please remove this section if you linked an issue above) -->
There is a wide variety of configurations that you can set and currently it is nontrivial for a client application to obtain listener information based on the configs. Writing the addresses to the status field is a handy choice and ease the integration of the clients to the cluster.

### 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

### TODO
- [x] Implement the actual status update
- [x] Test it on a remote cluster

adamantal: Tested on a remote cluster.
CR:
```yaml
spec:
  ...
  ingressController: envoy
  listenersConfig:
    externalListeners:
      - type: "plaintext"
        name: "external"
        externalStartingPort: 19090
        containerPort: 9094
    internalListeners:
      - type: "plaintext"
        name: "internal"
        containerPort: 29092
        usedForInnerBrokerCommunication: true
      - type: "plaintext"
        name: "controller"
        containerPort: 29093
        usedForInnerBrokerCommunication: false
        usedForControllerCommunication: true
```
and the status field:
```yaml
status:
  ...
  listenerStatuses:
    externalListeners:
      external:
        host: a3c22a3a170184e2c834b0e1f4eb3b29-711704928.eu-central-1.elb.amazonaws.com
        port: 9094
    internalListeners:
      internal:
        host: kafka-headless.kafka.svc.cluster.local
        port: 29092
```

Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index e27ca206..0be53c7b 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -8,7 +8,7 @@ jobs:
     resource_class: medium+
 
     docker:
-      - image: cimg/go:1.14
+      - image: cimg/go:1.15
 
     steps:
       - checkout
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 8aac6dd3..88806578 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -14,7 +14,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.14
+          go-version: 1.15
 
       - name: Checkout code
         uses: actions/checkout@v2
diff --git a/Dockerfile b/Dockerfile
index 58a0fd2e..260127f3 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,5 +1,5 @@
 # Build the manager binary
-FROM golang:1.14 as builder
+FROM golang:1.15 as builder
 
 WORKDIR /workspace
 # Copy the Go Modules manifests
diff --git a/charts/kafka-operator/templates/crds.yaml b/charts/kafka-operator/templates/crds.yaml
index f85858d0..b1df28e5 100644
--- a/charts/kafka-operator/templates/crds.yaml
+++ b/charts/kafka-operator/templates/crds.yaml
@@ -5056,6 +5056,10 @@ spec:
                       - cruiseControlState
                       - errorMessage
                     type: object
+                  perBrokerConfigurationState:
+                    description: PerBrokerConfigurationState holds info about the
+                      per-broker (dynamically updatable) config
+                    type: string
                   rackAwarenessState:
                     description: RackAwarenessState holds info about rack awareness
                       status
@@ -5063,6 +5067,7 @@ spec:
                 required:
                   - configurationState
                   - gracefulActionState
+                  - perBrokerConfigurationState
                   - rackAwarenessState
                 type: object
               type: object
@@ -5070,6 +5075,48 @@ spec:
               description: CruiseControlTopicStatus holds info about the CC topic
                 status
               type: string
+            listenerStatuses:
+              description: ListenerStatuses holds information about the statuses of
+                the configured listeners. The internal and external listeners are
+                stored in separate maps, and each listener can be looked up by name.
+              properties:
+                externalListeners:
+                  additionalProperties:
+                    description: ListenerStatusList can hold various amount of statuses
+                      based on the listener configuration.
+                    items:
+                      description: ListenerStatus holds information about the address
+                        of the listener
+                      properties:
+                        address:
+                          type: string
+                        name:
+                          type: string
+                      required:
+                        - address
+                        - name
+                      type: object
+                    type: array
+                  type: object
+                internalListeners:
+                  additionalProperties:
+                    description: ListenerStatusList can hold various amount of statuses
+                      based on the listener configuration.
+                    items:
+                      description: ListenerStatus holds information about the address
+                        of the listener
+                      properties:
+                        address:
+                          type: string
+                        name:
+                          type: string
+                      required:
+                        - address
+                        - name
+                      type: object
+                    type: array
+                  type: object
+              type: object
             rollingUpgradeStatus:
               description: RollingUpgradeStatus defines status of rolling upgrade
               properties:
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index dfebf1ab..69989b3c 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -5076,6 +5076,48 @@ spec:
               description: CruiseControlTopicStatus holds info about the CC topic
                 status
               type: string
+            listenerStatuses:
+              description: ListenerStatuses holds information about the statuses of
+                the configured listeners. The internal and external listeners are
+                stored in separate maps, and each listener can be looked up by name.
+              properties:
+                externalListeners:
+                  additionalProperties:
+                    description: ListenerStatusList can hold various amount of statuses
+                      based on the listener configuration.
+                    items:
+                      description: ListenerStatus holds information about the address
+                        of the listener
+                      properties:
+                        address:
+                          type: string
+                        name:
+                          type: string
+                      required:
+                      - address
+                      - name
+                      type: object
+                    type: array
+                  type: object
+                internalListeners:
+                  additionalProperties:
+                    description: ListenerStatusList can hold various amount of statuses
+                      based on the listener configuration.
+                    items:
+                      description: ListenerStatus holds information about the address
+                        of the listener
+                      properties:
+                        address:
+                          type: string
+                        name:
+                          type: string
+                      required:
+                      - address
+                      - name
+                      type: object
+                    type: array
+                  type: object
+              type: object
             rollingUpgradeStatus:
               description: RollingUpgradeStatus defines status of rolling upgrade
               properties:
diff --git a/controllers/kafkacluster_controller.go b/controllers/kafkacluster_controller.go
index 3780afcc..d906e80f 100644
--- a/controllers/kafkacluster_controller.go
+++ b/controllers/kafkacluster_controller.go
@@ -164,6 +164,10 @@ func (r *KafkaClusterReconciler) Reconcile(request ctrl.Request) (ctrl.Result, e
 				log.V(1).Info("dynamically updated broker configuration hasn't propagated through yet")
 				// for exponential backoff
 				return ctrl.Result{}, err
+			case errorfactory.LoadBalancerIPNotReady:
+				return ctrl.Result{
+					RequeueAfter: time.Duration(30) * time.Second,
+				}, nil
 			default:
 				return requeueWithError(log, err.Error(), err)
 			}
diff --git a/go.mod b/go.mod
index 2e0f0718..3483e2b6 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/banzaicloud/kafka-operator
 
-go 1.14
+go 1.15
 
 require (
 	emperror.dev/errors v0.7.0
diff --git a/pkg/errorfactory/errorfactory.go b/pkg/errorfactory/errorfactory.go
index 090d6119..a753665f 100644
--- a/pkg/errorfactory/errorfactory.go
+++ b/pkg/errorfactory/errorfactory.go
@@ -73,6 +73,9 @@ type CruiseControlTaskFailure struct{ error }
 // PerBrokerConfigNotReady states that per-broker configurations has been updated for a broker
 type PerBrokerConfigNotReady struct{ error }
 
+// LoadBalancerIPNotReady states that the LoadBalancer IP is not yet created
+type LoadBalancerIPNotReady struct{ error }
+
 // New creates a new error factory error
 func New(t interface{}, err error, msg string, wrapArgs ...interface{}) error {
 	wrapped := errors.WrapIfWithDetails(err, msg, wrapArgs...)
@@ -115,6 +118,8 @@ func New(t interface{}, err error, msg string, wrapArgs ...interface{}) error {
 		return CruiseControlTaskFailure{wrapped}
 	case PerBrokerConfigNotReady:
 		return PerBrokerConfigNotReady{wrapped}
+	case LoadBalancerIPNotReady:
+		return LoadBalancerIPNotReady{wrapped}
 	}
 	return wrapped
 }
diff --git a/pkg/k8sutil/status.go b/pkg/k8sutil/status.go
index 35c24eee..428effd6 100644
--- a/pkg/k8sutil/status.go
+++ b/pkg/k8sutil/status.go
@@ -22,6 +22,7 @@ import (
 
 	"emperror.dev/errors"
 	"github.com/go-logr/logr"
+
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
@@ -30,6 +31,7 @@ import (
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	banzaicloudv1beta1 "github.com/banzaicloud/kafka-operator/api/v1beta1"
+	clientutil "github.com/banzaicloud/kafka-operator/pkg/util/client"
 )
 
 // IsAlreadyOwnedError checks if a controller already own the instance
@@ -254,3 +256,90 @@ func UpdateRollingUpgradeState(c client.Client, cluster *v1beta1.KafkaCluster, t
 	logger.Info("Rolling upgrade status updated", "status", timeStamp)
 	return nil
 }
+
+func UpdateListenerStatuses(c client.Client, ctx context.Context, cluster *v1beta1.KafkaCluster, logger logr.Logger, intListenerStatuses, extListenerStatuses map[string]v1beta1.ListenerStatusList) error {
+	typeMeta := cluster.TypeMeta
+
+	cluster.Status.ListenerStatuses = v1beta1.ListenerStatuses{
+		InternalListeners: intListenerStatuses,
+		ExternalListeners: extListenerStatuses,
+	}
+
+	err := c.Status().Update(ctx, cluster)
+	if apierrors.IsNotFound(err) {
+		err = c.Update(ctx, cluster)
+	}
+	if err != nil {
+		if !apierrors.IsConflict(err) {
+			return errors.WrapIf(err, "could not update listener status")
+		}
+		err := c.Get(ctx, types.NamespacedName{
+			Namespace: cluster.Namespace,
+			Name:      cluster.Name,
+		}, cluster)
+		if err != nil {
+			return errors.WrapIf(err, "could not get config for updating listener status")
+		}
+
+		cluster.Status.ListenerStatuses = v1beta1.ListenerStatuses{
+			InternalListeners: intListenerStatuses,
+			ExternalListeners: extListenerStatuses,
+		}
+
+		err = c.Status().Update(ctx, cluster)
+		if apierrors.IsNotFound(err) {
+			err = c.Update(ctx, cluster)
+		}
+		if err != nil {
+			return errors.WrapIf(err, "could not update listener statuses")
+		}
+	}
+	// update loses the typeMeta of the config that's used later when setting ownerrefs
+	cluster.TypeMeta = typeMeta
+	logger.Info("updated listener statuses")
+	return nil
+}
+
+func CreateInternalListenerStatuses(kafkaCluster *v1beta1.KafkaCluster) (map[string]v1beta1.ListenerStatusList, map[string]v1beta1.ListenerStatusList) {
+	intListenerStatuses := make(map[string]v1beta1.ListenerStatusList, len(kafkaCluster.Spec.ListenersConfig.InternalListeners))
+	controllerIntListenerStatuses := make(map[string]v1beta1.ListenerStatusList)
+
+	internalAddress := clientutil.GenerateKafkaAddressWithoutPort(kafkaCluster)
+	for _, iListener := range kafkaCluster.Spec.ListenersConfig.InternalListeners {
+		listenerStatusList := v1beta1.ListenerStatusList{}
+
+		// add headless or any broker address
+		name := "any-broker"
+		if kafkaCluster.Spec.HeadlessServiceEnabled {
+			name = "headless"
+		}
+		listenerStatusList = append(listenerStatusList, v1beta1.ListenerStatus{
+			Name:    name,
+			Address: fmt.Sprintf("%s:%d", internalAddress, iListener.ContainerPort),
+		})
+
+		// add addresses per broker
+		for _, broker := range kafkaCluster.Spec.Brokers {
+			var address string
+			if kafkaCluster.Spec.HeadlessServiceEnabled {
+				address = fmt.Sprintf("%s-%d.%s-headless.%s.svc.%s:%d", kafkaCluster.Name, broker.Id, kafkaCluster.Name,
+					kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), iListener.ContainerPort)
+			} else {
+				address = fmt.Sprintf("%s-%d.%s.svc.%s:%d", kafkaCluster.Name, broker.Id, kafkaCluster.Namespace,
+					kafkaCluster.Spec.GetKubernetesClusterDomain(), iListener.ContainerPort)
+			}
+			listenerStatusList = append(listenerStatusList, v1beta1.ListenerStatus{
+				Name:    fmt.Sprintf("broker-%d", broker.Id),
+				Address: address,
+			})
+		}
+
+		if iListener.UsedForControllerCommunication {
+			controllerIntListenerStatuses[iListener.Name] = listenerStatusList
+		} else {
+			intListenerStatuses[iListener.Name] = listenerStatusList
+		}
+	}
+
+	return intListenerStatuses, controllerIntListenerStatuses
+}
diff --git a/pkg/kafkaclient/config.go b/pkg/kafkaclient/config.go
index 880d1d6c..c623a812 100644
--- a/pkg/kafkaclient/config.go
+++ b/pkg/kafkaclient/config.go
@@ -16,12 +16,12 @@ package kafkaclient
 
 import (
 	"crypto/tls"
-	"fmt"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/pki"
-	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
-	"sigs.k8s.io/controller-runtime/pkg/client"
+	clientutil "github.com/banzaicloud/kafka-operator/pkg/util/client"
 )
 
 const kafkaDefaultTimeout = int64(5)
@@ -38,9 +38,9 @@ type KafkaConfig struct {
 // ClusterConfig creates connection options from a KafkaCluster CR
 func ClusterConfig(client client.Client, cluster *v1beta1.KafkaCluster) (*KafkaConfig, error) {
 	conf := &KafkaConfig{}
-	conf.BrokerURI = generateKafkaAddress(cluster)
+	conf.BrokerURI = clientutil.GenerateKafkaAddress(cluster)
 	conf.OperationTimeout = kafkaDefaultTimeout
-	if cluster.Spec.ListenersConfig.SSLSecrets != nil && useSSL(cluster) {
+	if cluster.Spec.ListenersConfig.SSLSecrets != nil && clientutil.UseSSL(cluster) {
 		tlsConfig, err := pki.GetPKIManager(client, cluster, v1beta1.PKIBackendProvided).GetControllerTLSConfig()
 		if err != nil {
 			return conf, err
@@ -50,33 +50,3 @@ func ClusterConfig(client client.Client, cluster *v1beta1.KafkaCluster) (*KafkaC
 	}
 	return conf, nil
 }
-
-func useSSL(cluster *v1beta1.KafkaCluster) bool {
-	return cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].Type != "plaintext"
-}
-
-func determineInternalListenerForInnerCom(internalListeners []v1beta1.InternalListenerConfig) int {
-	for id, val := range internalListeners {
-		if val.UsedForInnerBrokerCommunication {
-			return id
-		}
-	}
-	return 0
-}
-
-func generateKafkaAddress(cluster *v1beta1.KafkaCluster) string {
-	if cluster.Spec.HeadlessServiceEnabled {
-		return fmt.Sprintf("%s.%s.svc.%s:%d",
-			fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name),
-			cluster.Namespace,
-			cluster.Spec.GetKubernetesClusterDomain(),
-			cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].ContainerPort,
-		)
-	}
-	return fmt.Sprintf("%s.%s.svc.%s:%d",
-		fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name),
-		cluster.Namespace,
-		cluster.Spec.GetKubernetesClusterDomain(),
-		cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].ContainerPort,
-	)
-}
diff --git a/pkg/pki/certmanagerpki/certmanager_pki.go b/pkg/pki/certmanagerpki/certmanager_pki.go
index 8a606053..23038cbe 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki.go
@@ -90,10 +90,10 @@ func (c *certManager) FinalizePKI(ctx context.Context, logger logr.Logger) error
 	return nil
 }
 
-func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) (err error) {
+func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) (err error) {
 	logger.Info("Reconciling cert-manager PKI")
 
-	resources, err := c.kafkapki(ctx, scheme, externalHostnames)
+	resources, err := c.kafkapki(ctx, scheme, extListenerStatuses)
 	if err != nil {
 		return err
 	}
@@ -107,28 +107,28 @@ func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, sche
 	return nil
 }
 
-func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, externalHostnames map[string]string) ([]runtime.Object, error) {
+func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) ([]runtime.Object, error) {
 	sslConfig := c.cluster.Spec.ListenersConfig.SSLSecrets
 	if sslConfig.Create {
 		if sslConfig.IssuerRef == nil {
-			return fullPKI(c.cluster, scheme, externalHostnames), nil
+			return fullPKI(c.cluster, scheme, extListenerStatuses), nil
 		}
-		return userProvidedIssuerPKI(c.cluster, externalHostnames), nil
+		return userProvidedIssuerPKI(c.cluster, extListenerStatuses), nil
 	}
-	return userProvidedPKI(ctx, c.client, c.cluster, scheme, externalHostnames)
+	return userProvidedPKI(ctx, c.client, c.cluster, scheme, extListenerStatuses)
 }
 
-func userProvidedIssuerPKI(cluster *v1beta1.KafkaCluster, externalHostnames map[string]string) []runtime.Object {
+func userProvidedIssuerPKI(cluster *v1beta1.KafkaCluster, extListenerStatuses map[string]v1beta1.ListenerStatusList) []runtime.Object {
 	// No need to generate self-signed certs and issuers because the issuer is provided by user
 	return []runtime.Object{
 		// Broker "user"
-		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
+		pkicommon.BrokerUserForCluster(cluster, extListenerStatuses),
 		// Operator user
 		pkicommon.ControllerUserForCluster(cluster),
 	}
 }
 
-func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames map[string]string) []runtime.Object {
+func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) []runtime.Object {
 	return []runtime.Object{
 		// A self-signer for the CA Certificate
 		selfSignerForCluster(cluster, scheme),
@@ -138,7 +138,7 @@ func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHost
 		// for producers/consumers in other namespaces
 		mainIssuerForCluster(cluster),
 		// Broker "user"
-		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
+		pkicommon.BrokerUserForCluster(cluster, extListenerStatuses),
 		// Operator user
 		pkicommon.ControllerUserForCluster(cluster),
 	}
@@ -146,7 +146,7 @@ func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHost
 
 func userProvidedPKI(
 	ctx context.Context, client client.Client,
-	cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames map[string]string) ([]runtime.Object, error) {
+	cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) ([]runtime.Object, error) {
 	// If we aren't creating the secrets we need a cluster issuer made from the provided secret
 	caSecret, err := caSecretForProvidedCert(ctx, client, cluster, scheme)
 	if err != nil {
@@ -161,7 +161,7 @@ func userProvidedPKI(
 		//
 		// TODO: (tinyzimmer) - Would it be better to allow the KafkaUser to take a user-provided cert/key combination?
 		// It would have to be validated first as signed by whatever the CA is - probably via a webhook.
-		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
+		pkicommon.BrokerUserForCluster(cluster, extListenerStatuses),
 		pkicommon.ControllerUserForCluster(cluster),
 	}, nil
 }
diff --git a/pkg/pki/pki_manager.go b/pkg/pki/pki_manager.go
index b3c4e76c..2102390c 100644
--- a/pkg/pki/pki_manager.go
+++ b/pkg/pki/pki_manager.go
@@ -72,7 +72,7 @@ func newMockPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pki.
 	return &mockPKIManager{client: client, cluster: cluster}
 }
 
-func (m *mockPKIManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) error {
+func (m *mockPKIManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) error {
 	return nil
 }
 
diff --git a/pkg/pki/vaultpki/vault_pki.go b/pkg/pki/vaultpki/vault_pki.go
index b49f3f81..19f4327b 100644
--- a/pkg/pki/vaultpki/vault_pki.go
+++ b/pkg/pki/vaultpki/vault_pki.go
@@ -28,6 +28,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
@@ -56,7 +57,7 @@ func (v *vaultPKI) FinalizePKI(_ context.Context, _ logr.Logger) error {
 }
 
 // ReconcilePKI will use the user-provided paths to ensure broker/operator users for a new KafkaCluster
-func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) (err error) {
+func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, extListenerStatuses map[string]v1beta1.ListenerStatusList) (err error) {
 	log := logger.WithName("vault_pki")
 
 	log.Info("Retrieving vault client")
@@ -66,7 +67,7 @@ func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme
 	}
 
 	log.Info("Reconciling broker user in vault")
-	brokerCert, err := v.reconcileBrokerCert(ctx, vault, externalHostnames)
+	brokerCert, err := v.reconcileBrokerCert(ctx, vault, extListenerStatuses)
 	if err != nil {
 		return
 	}
@@ -153,8 +154,8 @@ func (v *vaultPKI) reconcileBootstrapSecrets(ctx context.Context, scheme *runtim
 }
 
 func (v *vaultPKI) reconcileBrokerCert(
-	ctx context.Context, vault *vaultapi.Client, externalHostnames map[string]string) (*pkicommon.UserCertificate, error) {
-	return v.reconcileStartupUser(ctx, vault, pkicommon.BrokerUserForCluster(v.cluster, externalHostnames))
+	ctx context.Context, vault *vaultapi.Client, extListenerStatuses map[string]v1beta1.ListenerStatusList) (*pkicommon.UserCertificate, error) {
+	return v.reconcileStartupUser(ctx, vault, pkicommon.BrokerUserForCluster(v.cluster, extListenerStatuses))
 }
 
 func (v *vaultPKI) reconcileControllerCert(ctx context.Context, vault *vaultapi.Client) (*pkicommon.UserCertificate, error) {
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index ab89e0e8..ab39841e 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -77,7 +77,7 @@ super.users={{ .SuperUsers }}
 {{ end }}
 `
 
-func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, loadBalancerIPs map[string]string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
 	var out bytes.Buffer
 	t := template.Must(template.New("bConfig-config").Parse(kafkaConfigTemplate))
 	if err := t.Execute(&out, map[string]interface{}{
@@ -88,18 +88,15 @@ func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, lo
 		"ZookeeperConnectString":             zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath()),
 		"CruiseControlBootstrapServers":      getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
 		"StorageConfig":                      generateStorageConfig(bConfig.StorageConfigs),
-		"AdvertisedListenersConfig": generateAdvertisedListenerConfig(id,
-			r.KafkaCluster.Spec.ListenersConfig, loadBalancerIPs, r.KafkaCluster.Spec.GetKubernetesClusterDomain(),
-			r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled,
-			bConfig),
-		"SuperUsers":             strings.Join(generateSuperUsers(superUsers), ";"),
-		"ServerKeystorePath":     serverKeystorePath,
-		"ClientKeystorePath":     clientKeystorePath,
-		"KeyStoreFile":           v1alpha1.TLSJKSKeyStore,
-		"TrustStoreFile":         v1alpha1.TLSJKSTrustStore,
-		"ServerKeystorePassword": serverPass,
-		"ClientKeystorePassword": clientPass,
-		"ControlPlaneListener":   generateControlPlaneListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
+		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses),
+		"SuperUsers":                         strings.Join(generateSuperUsers(superUsers), ";"),
+		"ServerKeystorePath":                 serverKeystorePath,
+		"ClientKeystorePath":                 clientKeystorePath,
+		"KeyStoreFile":                       v1alpha1.TLSJKSKeyStore,
+		"TrustStoreFile":                     v1alpha1.TLSJKSTrustStore,
+		"ServerKeystorePassword":             serverPass,
+		"ClientKeystorePassword":             clientPass,
+		"ControlPlaneListener":               generateControlPlaneListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
 	}); err != nil {
 		log.Error(err, "error occurred during parsing the config template")
 	}
@@ -114,7 +111,7 @@ func generateSuperUsers(users []string) (suStrings []string) {
 	return
 }
 
-func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs map[string]string, serverPass, clientPass string, superUsers []string, log logr.Logger) *corev1.ConfigMap {
+func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) *corev1.ConfigMap {
 	brokerConf := &corev1.ConfigMap{
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(brokerConfigTemplate+"-%d", r.KafkaCluster.Name, id),
@@ -124,7 +121,7 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loa
 			),
 			r.KafkaCluster,
 		),
-		Data: map[string]string{kafka.ConfigPropertyName: r.generateBrokerConfig(id, brokerConfig, loadBalancerIPs, serverPass, clientPass, superUsers, log)},
+		Data: map[string]string{kafka.ConfigPropertyName: r.generateBrokerConfig(id, brokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)},
 	}
 	if brokerConfig.Log4jConfig != "" {
 		brokerConf.Data["log4j.properties"] = brokerConfig.Log4jConfig
@@ -133,42 +130,27 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loa
 }
 
 func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig,
-	loadBalancerIPs map[string]string, domain, namespace, crName string, headlessServiceEnabled bool, bConfig *v1beta1.BrokerConfig) string {
+	extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList) string {
 	advertisedListenerConfig := make([]string, 0, len(l.ExternalListeners)+len(l.InternalListeners))
-	for _, eListener := range l.ExternalListeners {
-		if eListener.GetAccessMethod() == corev1.ServiceTypeLoadBalancer {
-			advertisedListenerConfig = append(advertisedListenerConfig,
-				fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), loadBalancerIPs[eListener.Name], eListener.ExternalStartingPort+id))
-		} else {
-			portNumber := eListener.ExternalStartingPort + id
-			if externalIP, ok := bConfig.NodePortExternalIP[eListener.Name]; ok && externalIP != "" {
-				// https://kubernetes.io/docs/concepts/services-networking/service/#external-ips
-				// if specific external IP is set for the NodePort service incoming traffic will be received on Service port
-				// and not nodeport
-				portNumber = eListener.ContainerPort
-			}
 
-			if _, ok := loadBalancerIPs[eListener.Name]; !ok {
-				nodePortExternalIP := bConfig.NodePortExternalIP[eListener.Name]
-				advertisedListenerConfig = append(advertisedListenerConfig,
-					fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), nodePortExternalIP, portNumber))
-			} else {
+	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, extListenerStatuses)
+	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, intListenerStatuses)
+	advertisedListenerConfig = appendListenerConfigs(advertisedListenerConfig, id, controllerIntListenerStatuses)
+
+	return fmt.Sprintf("advertised.listeners=%s\n", strings.Join(advertisedListenerConfig, ","))
+}
+
+func appendListenerConfigs(advertisedListenerConfig []string, id int32, listenerStatusList map[string]v1beta1.ListenerStatusList) []string {
+	for iListenerName, statuses := range listenerStatusList {
+		for _, status := range statuses {
+			if status.Name == fmt.Sprintf("broker-%d", id) {
 				advertisedListenerConfig = append(advertisedListenerConfig,
-					fmt.Sprintf("%s://%s-%d-%s.%s%s:%d", strings.ToUpper(eListener.Name), crName, id, eListener.Name, namespace, loadBalancerIPs[eListener.Name], portNumber))
+					fmt.Sprintf("%s://%s", strings.ToUpper(iListenerName), status.Address))
+				break
 			}
-
-		}
-	}
-	for _, iListener := range l.InternalListeners {
-		if headlessServiceEnabled {
-			advertisedListenerConfig = append(advertisedListenerConfig,
-				fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, domain, iListener.ContainerPort))
-		} else {
-			advertisedListenerConfig = append(advertisedListenerConfig,
-				fmt.Sprintf("%s://%s-%d.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, namespace, domain, iListener.ContainerPort))
 		}
 	}
-	return fmt.Sprintf("advertised.listeners=%s\n", strings.Join(advertisedListenerConfig, ","))
+	return advertisedListenerConfig
 }
 
 func generateStorageConfig(sConfig []v1beta1.StorageConfig) string {
@@ -238,7 +220,7 @@ func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32,
 	return internalListener
 }
 
-func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs map[string]string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses map[string]v1beta1.ListenerStatusList, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
 	parsedReadOnlyClusterConfig := util.ParsePropertiesFormat(r.KafkaCluster.Spec.ReadOnlyConfig)
 	var parsedReadOnlyBrokerConfig = map[string]string{}
 
@@ -256,7 +238,7 @@ func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerC
 	//Generate the Complete Configuration for the Broker
 	completeConfigMap := map[string]string{}
 
-	if err := mergo.Merge(&completeConfigMap, util.ParsePropertiesFormat(r.getConfigString(brokerConfig, id, loadBalancerIPs, serverPass, clientPass, superUsers, log))); err != nil {
+	if err := mergo.Merge(&completeConfigMap, util.ParsePropertiesFormat(r.getConfigString(brokerConfig, id, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log))); err != nil {
 		log.Error(err, "error occurred during merging operator generated configs")
 	}
 
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index 337ba4e8..dfd04529 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -17,15 +17,16 @@ package kafka
 import (
 	"context"
 	"fmt"
+	envoyutils "github.com/banzaicloud/kafka-operator/pkg/util/envoy"
 	"reflect"
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 
 	"emperror.dev/errors"
 	"github.com/banzaicloud/k8s-objectmatcher/patch"
 	"github.com/go-logr/logr"
+
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -44,7 +45,6 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/scale"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	certutil "github.com/banzaicloud/kafka-operator/pkg/util/cert"
-	envoyutils "github.com/banzaicloud/kafka-operator/pkg/util/envoy"
 	istioingressutils "github.com/banzaicloud/kafka-operator/pkg/util/istioingress"
 	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
 	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
@@ -108,27 +108,13 @@ func getCreatedPvcForBroker(c client.Client, brokerID int32, namespace, crName s
 	return foundPvcList.Items, nil
 }
 
-func getLoadBalancerIP(client client.Client, namespace, ingressController, crName, extListenerName string) (string, error) {
-	foundLBService := &corev1.Service{}
-	var iControllerServiceName string
-	if ingressController == istioingressutils.IngressControllerName {
-		iControllerServiceName = fmt.Sprintf(istioingressutils.MeshGatewayNameTemplate, extListenerName, crName)
-	} else if ingressController == envoyutils.IngressControllerName {
-		iControllerServiceName = fmt.Sprintf(envoyutils.EnvoyServiceName, extListenerName, crName)
-	}
-
-	err := client.Get(context.TODO(), types.NamespacedName{Name: iControllerServiceName, Namespace: namespace}, foundLBService)
-	if err != nil {
-		return "", err
-	}
-
+func getLoadBalancerIP(foundLBService *corev1.Service) (string, error) {
 	if len(foundLBService.Status.LoadBalancer.Ingress) == 0 {
 		return "", errorfactory.New(errorfactory.ResourceNotReady{}, errors.New("loadbalancer is not created waiting"), "trying")
 	}
 
 	if foundLBService.Status.LoadBalancer.Ingress[0].Hostname == "" && foundLBService.Status.LoadBalancer.Ingress[0].IP == "" {
-		time.Sleep(20 * time.Second)
-		return "", errorfactory.New(errorfactory.ResourceNotReady{}, errors.New("loadbalancer is not created waiting"), "trying")
+		return "", errorfactory.New(errorfactory.LoadBalancerIPNotReady{}, errors.New("loadbalancer hostname and IP has not been created yet - waiting"), "trying")
 	}
 	var loadBalancerExternalAddress string
 	if foundLBService.Status.LoadBalancer.Ingress[0].Hostname == "" {
@@ -192,25 +178,20 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		return errors.WrapIf(err, "failed to reconcile resource")
 	}
 
-	lbIPs := make(map[string]string, len(r.KafkaCluster.Spec.ListenersConfig.ExternalListeners))
-
-	for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
-		if eListener.HostnameOverride != "" {
-			lbIPs[eListener.Name] = eListener.HostnameOverride
-		} else if eListener.GetAccessMethod() == corev1.ServiceTypeLoadBalancer {
-			lbIP, err := getLoadBalancerIP(r.Client, r.KafkaCluster.GetNamespace(),
-				r.KafkaCluster.Spec.GetIngressController(), r.KafkaCluster.GetName(), eListener.Name)
-			if err != nil {
-				return err
-			}
-			lbIPs[eListener.Name] = lbIP
-		}
+	extListenerStatuses, err := r.createExternalListenerStatuses()
+	if err != nil {
+		return errors.WrapIf(err, "could not update status for external listeners")
+	}
+	intListenerStatuses, controllerIntListenerStatuses := k8sutil.CreateInternalListenerStatuses(r.KafkaCluster)
+	err = k8sutil.UpdateListenerStatuses(r.Client, context.Background(), r.KafkaCluster, log, intListenerStatuses, extListenerStatuses)
+	if err != nil {
+		return errors.WrapIf(err, "failed to update listener statuses")
 	}
 
 	// Setup the PKI if using SSL
 	if r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil {
 		// reconcile the PKI
-		if err := pki.GetPKIManager(r.Client, r.KafkaCluster, v1beta1.PKIBackendProvided).ReconcilePKI(context.TODO(), log, r.Scheme, lbIPs); err != nil {
+		if err := pki.GetPKIManager(r.Client, r.KafkaCluster, v1beta1.PKIBackendProvided).ReconcilePKI(context.TODO(), log, r.Scheme, extListenerStatuses); err != nil {
 			return err
 		}
 	}
@@ -253,7 +234,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 		var configMap *corev1.ConfigMap
 		if r.KafkaCluster.Spec.RackAwareness == nil {
-			configMap = r.configMap(broker.Id, brokerConfig, lbIPs, serverPass, clientPass, superUsers, log)
+			configMap = r.configMap(broker.Id, brokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)
 			err := k8sutil.Reconcile(log, r.Client, configMap, r.KafkaCluster)
 			if err != nil {
 				return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", configMap.GetObjectKind().GroupVersionKind())
@@ -261,7 +242,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		} else {
 			if brokerState, ok := r.KafkaCluster.Status.BrokersState[strconv.Itoa(int(broker.Id))]; ok {
 				if brokerState.RackAwarenessState != "" {
-					configMap = r.configMap(broker.Id, brokerConfig, lbIPs, serverPass, clientPass, superUsers, log)
+					configMap = r.configMap(broker.Id, brokerConfig, extListenerStatuses, intListenerStatuses, controllerIntListenerStatuses, serverPass, clientPass, superUsers, log)
 					err := k8sutil.Reconcile(log, r.Client, configMap, r.KafkaCluster)
 					if err != nil {
 						return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", configMap.GetObjectKind().GroupVersionKind())
@@ -804,3 +785,102 @@ func GetBrokersWithPendingOrRunningCCTask(kafkaCluster *v1beta1.KafkaCluster) []
 func isDesiredStorageValueInvalid(desired, current *corev1.PersistentVolumeClaim) bool {
 	return desired.Spec.Resources.Requests.Storage().Value() < current.Spec.Resources.Requests.Storage().Value()
 }
+
+func (r *Reconciler) createExternalListenerStatuses() (map[string]v1beta1.ListenerStatusList, error) {
+	extListenerStatuses := make(map[string]v1beta1.ListenerStatusList, len(r.KafkaCluster.Spec.ListenersConfig.ExternalListeners))
+	for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
+		var host string
+		var foundLBService *corev1.Service
+		var err error
+
+		if eListener.HostnameOverride != "" {
+			host = eListener.HostnameOverride
+		} else if eListener.GetAccessMethod() == corev1.ServiceTypeLoadBalancer {
+			foundLBService, err = getServiceFromExternalListener(r.Client, r.KafkaCluster, eListener.Name)
+			if err != nil {
+				return nil, errors.WrapIfWithDetails(err, "could not get service corresponding to the external listener", "externalListenerName", eListener.Name)
+			}
+			lbIP, err := getLoadBalancerIP(foundLBService)
+			if err != nil {
+				return nil, errors.WrapIfWithDetails(err, "could not extract IP from LoadBalancer service", "externalListenerName", eListener.Name)
+			}
+			host = lbIP
+		}
+		listenerStatusList := make(v1beta1.ListenerStatusList, 0, len(r.KafkaCluster.Spec.Brokers)+1)
+
+		// optionally add all brokers service to the top of the list
+		if eListener.GetAccessMethod() != corev1.ServiceTypeNodePort {
+			if foundLBService == nil {
+				foundLBService, err = getServiceFromExternalListener(r.Client, r.KafkaCluster, eListener.Name)
+				if err != nil {
+					return nil, errors.WrapIfWithDetails(err, "could not get service corresponding to the external listener", "externalListenerName", eListener.Name)
+				}
+			}
+			var allBrokerPort int32 = 0
+			for _, port := range foundLBService.Spec.Ports {
+				if port.Name == "tcp-all-broker" {
+					allBrokerPort = port.Port
+					break
+				}
+			}
+			if allBrokerPort == 0 {
+				return nil, errors.NewWithDetails("could not find port with name tcp-all-broker", "externalListenerName", eListener.Name)
+			}
+			listenerStatus := v1beta1.ListenerStatus{
+				Name:    "any-broker",
+				Address: fmt.Sprintf("%s:%d", host, allBrokerPort),
+			}
+			listenerStatusList = append(listenerStatusList, listenerStatus)
+		}
+
+		for _, broker := range r.KafkaCluster.Spec.Brokers {
+			brokerHost := host
+			portNumber := eListener.ExternalStartingPort + broker.Id
+
+			if eListener.GetAccessMethod() != corev1.ServiceTypeLoadBalancer {
+				bConfig, err := util.GetBrokerConfig(broker, r.KafkaCluster.Spec)
+				if err != nil {
+					return nil, err
+				}
+
+				if externalIP, ok := bConfig.NodePortExternalIP[eListener.Name]; ok && externalIP != "" {
+					// https://kubernetes.io/docs/concepts/services-networking/service/#external-ips
+					// if specific external IP is set for the NodePort service incoming traffic will be received on Service port
+					// and not nodeport
+					portNumber = eListener.ContainerPort
+				}
+				if brokerHost == "" {
+					brokerHost = bConfig.NodePortExternalIP[eListener.Name]
+				} else {
+					brokerHost = fmt.Sprintf("%s-%d-%s.%s%s", r.KafkaCluster.Name, broker.Id, eListener.Name, r.KafkaCluster.Namespace, brokerHost)
+				}
+			}
+
+			listenerStatus := v1beta1.ListenerStatus{
+				Name:    fmt.Sprintf("broker-%d", broker.Id),
+				Address: fmt.Sprintf("%s:%d", brokerHost, portNumber),
+			}
+			listenerStatusList = append(listenerStatusList, listenerStatus)
+		}
+
+		extListenerStatuses[eListener.Name] = listenerStatusList
+	}
+	return extListenerStatuses, nil
+}
+
+func getServiceFromExternalListener(client client.Client, cluster *v1beta1.KafkaCluster, eListenerName string) (*corev1.Service, error) {
+	foundLBService := &corev1.Service{}
+	var iControllerServiceName string
+	switch cluster.Spec.GetIngressController() {
+	case istioingressutils.IngressControllerName:
+		iControllerServiceName = fmt.Sprintf(istioingressutils.MeshGatewayNameTemplate, eListenerName, cluster.GetName())
+	case envoyutils.IngressControllerName:
+		iControllerServiceName = fmt.Sprintf(envoyutils.EnvoyServiceName, eListenerName, cluster.GetName())
+	}
+
+	err := client.Get(context.TODO(), types.NamespacedName{Name: iControllerServiceName, Namespace: cluster.GetNamespace()}, foundLBService)
+	if err != nil {
+		return nil, errors.WrapIfWithDetails(err, "could not get LoadBalancer service", "serviceName", iControllerServiceName)
+	}
+	return foundLBService, nil
+}
diff --git a/pkg/sdk/go.mod b/pkg/sdk/go.mod
index f599928e..be213469 100644
--- a/pkg/sdk/go.mod
+++ b/pkg/sdk/go.mod
@@ -1,6 +1,6 @@
 module github.com/banzaicloud/kafka-operator/pkg/sdk
 
-go 1.14
+go 1.15
 
 require (
 	github.com/banzaicloud/istio-client-go v0.0.9
diff --git a/pkg/sdk/v1beta1/kafkacluster_types.go b/pkg/sdk/v1beta1/kafkacluster_types.go
index ff9a3726..2e02a6ac 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types.go
@@ -74,6 +74,7 @@ type KafkaClusterStatus struct {
 	State                    ClusterState             `json:"state"`
 	RollingUpgrade           RollingUpgradeStatus     `json:"rollingUpgradeStatus,omitempty"`
 	AlertCount               int                      `json:"alertCount"`
+	ListenerStatuses         ListenerStatuses         `json:"listenerStatuses,omitempty"`
 }
 
 // RollingUpgradeStatus defines status of rolling upgrade
@@ -363,6 +364,22 @@ type CommonListenerSpec struct {
 	ContainerPort int32  `json:"containerPort"`
 }
 
+// ListenerStatuses holds information about the statuses of the configured listeners.
+// The internal and external listeners are stored in separate maps, and each listener can be looked up by name.
+type ListenerStatuses struct {
+	InternalListeners map[string]ListenerStatusList `json:"internalListeners,omitempty"`
+	ExternalListeners map[string]ListenerStatusList `json:"externalListeners,omitempty"`
+}
+
+// ListenerStatusList can hold various amount of statuses based on the listener configuration.
+type ListenerStatusList []ListenerStatus
+
+// ListenerStatus holds information about the address of the listener
+type ListenerStatus struct {
+	Name    string `json:"name"`
+	Address string `json:"address"`
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:printcolumn:JSONPath=".status.state",name="Cluster state",type="string"
diff --git a/pkg/sdk/v1beta1/zz_generated.deepcopy.go b/pkg/sdk/v1beta1/zz_generated.deepcopy.go
index c2ebe5e8..96dc06ca 100644
--- a/pkg/sdk/v1beta1/zz_generated.deepcopy.go
+++ b/pkg/sdk/v1beta1/zz_generated.deepcopy.go
@@ -563,6 +563,7 @@ func (in *KafkaClusterStatus) DeepCopyInto(out *KafkaClusterStatus) {
 		}
 	}
 	out.RollingUpgrade = in.RollingUpgrade
+	in.ListenerStatuses.DeepCopyInto(&out.ListenerStatuses)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaClusterStatus.
@@ -575,6 +576,85 @@ func (in *KafkaClusterStatus) DeepCopy() *KafkaClusterStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ListenerStatus) DeepCopyInto(out *ListenerStatus) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ListenerStatus.
+func (in *ListenerStatus) DeepCopy() *ListenerStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(ListenerStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in ListenerStatusList) DeepCopyInto(out *ListenerStatusList) {
+	{
+		in := &in
+		*out = make(ListenerStatusList, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ListenerStatusList.
+func (in ListenerStatusList) DeepCopy() ListenerStatusList {
+	if in == nil {
+		return nil
+	}
+	out := new(ListenerStatusList)
+	in.DeepCopyInto(out)
+	return *out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ListenerStatuses) DeepCopyInto(out *ListenerStatuses) {
+	*out = *in
+	if in.InternalListeners != nil {
+		in, out := &in.InternalListeners, &out.InternalListeners
+		*out = make(map[string]ListenerStatusList, len(*in))
+		for key, val := range *in {
+			var outVal []ListenerStatus
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(ListenerStatusList, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+	}
+	if in.ExternalListeners != nil {
+		in, out := &in.ExternalListeners, &out.ExternalListeners
+		*out = make(map[string]ListenerStatusList, len(*in))
+		for key, val := range *in {
+			var outVal []ListenerStatus
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make(ListenerStatusList, len(*in))
+				copy(*out, *in)
+			}
+			(*out)[key] = outVal
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ListenerStatuses.
+func (in *ListenerStatuses) DeepCopy() *ListenerStatuses {
+	if in == nil {
+		return nil
+	}
+	out := new(ListenerStatuses)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ListenersConfig) DeepCopyInto(out *ListenersConfig) {
 	*out = *in
diff --git a/pkg/util/client/common.go b/pkg/util/client/common.go
new file mode 100644
index 00000000..1cf41163
--- /dev/null
+++ b/pkg/util/client/common.go
@@ -0,0 +1,55 @@
+// Copyright © 2020 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 client
+
+import (
+	"fmt"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
+)
+
+func UseSSL(cluster *v1beta1.KafkaCluster) bool {
+	return cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].Type != "plaintext"
+}
+
+func determineInternalListenerForInnerCom(internalListeners []v1beta1.InternalListenerConfig) int {
+	for id, val := range internalListeners {
+		if val.UsedForInnerBrokerCommunication {
+			return id
+		}
+	}
+	return 0
+}
+
+func GenerateKafkaAddressWithoutPort(cluster *v1beta1.KafkaCluster) string {
+	if cluster.Spec.HeadlessServiceEnabled {
+		return fmt.Sprintf("%s.%s.svc.%s",
+			fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name),
+			cluster.Namespace,
+			cluster.Spec.GetKubernetesClusterDomain(),
+		)
+	}
+	return fmt.Sprintf("%s.%s.svc.%s",
+		fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name),
+		cluster.Namespace,
+		cluster.Spec.GetKubernetesClusterDomain(),
+	)
+}
+
+func GenerateKafkaAddress(cluster *v1beta1.KafkaCluster) string {
+	return fmt.Sprintf("%s:%d", GenerateKafkaAddressWithoutPort(cluster),
+		cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].ContainerPort)
+}
diff --git a/pkg/util/pki/common.go b/pkg/util/pki/common.go
index 86bfee51..cdff6ad6 100644
--- a/pkg/util/pki/common.go
+++ b/pkg/util/pki/common.go
@@ -19,6 +19,7 @@ import (
 	"crypto/tls"
 	"fmt"
 	"sort"
+	"strings"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -55,7 +56,7 @@ type Manager interface {
 	// ReconcilePKI ensures a PKI for a kafka cluster - should be idempotent.
 	// This method should at least setup any issuer needed for user certificates
 	// as well as broker/cruise-control secrets
-	ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) error
+	ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]v1beta1.ListenerStatusList) error
 
 	// FinalizePKI performs any cleanup steps necessary for a PKI backend
 	FinalizePKI(ctx context.Context, logger logr.Logger) error
@@ -170,12 +171,15 @@ func LabelsForKafkaPKI(name, namespace string) map[string]string {
 }
 
 // BrokerUserForCluster returns a KafkaUser CR for the broker certificates in a KafkaCluster
-func BrokerUserForCluster(cluster *v1beta1.KafkaCluster, additionalHostnames map[string]string) *v1alpha1.KafkaUser {
-	additionalHosts := make([]string, 0, len(additionalHostnames))
-	for _, hostnames := range additionalHostnames {
-		additionalHosts = append(additionalHosts, hostnames)
+func BrokerUserForCluster(cluster *v1beta1.KafkaCluster, extListenerStatuses map[string]v1beta1.ListenerStatusList) *v1alpha1.KafkaUser {
+	additionalHosts := make([]string, 0, len(extListenerStatuses))
+	for _, listenerStatus := range extListenerStatuses {
+		for _, status := range listenerStatus {
+			host := strings.Split(status.Address, ":")[0]
+			additionalHosts = append(additionalHosts, host)
+		}
 	}
-	sort.Strings(additionalHosts)
+	additionalHosts = sortAndDedupe(additionalHosts)
 	return &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster), LabelsForKafkaPKI(cluster.Name, cluster.Namespace), cluster),
 		Spec: v1alpha1.KafkaUserSpec{
@@ -190,6 +194,20 @@ func BrokerUserForCluster(cluster *v1beta1.KafkaCluster, additionalHostnames map
 	}
 }
 
+func sortAndDedupe(hosts []string) []string {
+	sort.Strings(hosts)
+
+	uniqueHosts := make([]string, 0)
+	lastHost := ""
+	for _, host := range hosts {
+		if lastHost != host {
+			uniqueHosts = append(uniqueHosts, host)
+		}
+		lastHost = host
+	}
+	return uniqueHosts
+}
+
 // ControllerUserForCluster returns a KafkaUser CR for the controller/cc certificates in a KafkaCluster
 func ControllerUserForCluster(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaUser {
 	return &v1alpha1.KafkaUser{


Test patch:
diff --git a/controllers/tests/common_test.go b/controllers/tests/common_test.go
index 77f6b9dd..feb1e906 100644
--- a/controllers/tests/common_test.go
+++ b/controllers/tests/common_test.go
@@ -41,9 +41,9 @@ func createMinimalKafkaClusterCR(name, namespace string) *v1beta1.KafkaCluster {
 					{
 						CommonListenerSpec: v1beta1.CommonListenerSpec{
 							Name:          "test",
-							ContainerPort: 9733,
+							ContainerPort: 9094,
 						},
-						ExternalStartingPort: 11202,
+						ExternalStartingPort: 19090,
 						HostnameOverride:     "test-host",
 						AccessMethod:         corev1.ServiceTypeLoadBalancer,
 					},
@@ -92,6 +92,14 @@ func createMinimalKafkaClusterCR(name, namespace string) *v1beta1.KafkaCluster {
 					Id:                0,
 					BrokerConfigGroup: "default",
 				},
+				{
+					Id:                1,
+					BrokerConfigGroup: "default",
+				},
+				{
+					Id:                2,
+					BrokerConfigGroup: "default",
+				},
 			},
 			ClusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1",
 			ZKAddresses:  []string{},
diff --git a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
index 508616d9..22d23db0 100644
--- a/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
+++ b/controllers/tests/kafkacluster_controller_cruisecontrol_test.go
@@ -129,6 +129,30 @@ zookeeper.connect=/
                 "NW_OUT": "125000"
             },
             "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+        },
+        {
+            "brokerId": "1",
+            "capacity": {
+                "DISK": {
+                    "/kafka-logs/kafka": "10737418240"
+                },
+                "CPU": "150",
+                "NW_IN": "125000",
+                "NW_OUT": "125000"
+            },
+            "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
+        },
+        {
+            "brokerId": "2",
+            "capacity": {
+                "DISK": {
+                    "/kafka-logs/kafka": "10737418240"
+                },
+                "CPU": "150",
+                "NW_IN": "125000",
+                "NW_OUT": "125000"
+            },
+            "doc": "Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
         }
     ]
 }`))
diff --git a/controllers/tests/kafkacluster_controller_envoy_test.go b/controllers/tests/kafkacluster_controller_envoy_test.go
index acc1878b..3eccd749 100644
--- a/controllers/tests/kafkacluster_controller_envoy_test.go
+++ b/controllers/tests/kafkacluster_controller_envoy_test.go
@@ -17,7 +17,6 @@ package tests
 import (
 	"context"
 	"fmt"
-
 	. "github.com/onsi/gomega"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -49,15 +48,20 @@ func expectEnvoy(kafkaCluster *v1beta1.KafkaCluster) {
 		"eListenerName": "test",
 		"kafka_cr":      kafkaCluster.Name,
 	}))
-	Expect(loadBalancer.Spec.Ports).To(HaveLen(2))
-	Expect(loadBalancer.Spec.Ports[0].Name).To(Equal("broker-0"))
-	Expect(loadBalancer.Spec.Ports[0].Protocol).To(Equal(corev1.ProtocolTCP))
-	Expect(loadBalancer.Spec.Ports[0].Port).To(BeEquivalentTo(11202))
-	Expect(loadBalancer.Spec.Ports[0].TargetPort.IntVal).To(BeEquivalentTo(11202))
-	Expect(loadBalancer.Spec.Ports[1].Name).To(Equal("tcp-all-broker"))
-	Expect(loadBalancer.Spec.Ports[1].Protocol).To(Equal(corev1.ProtocolTCP))
-	Expect(loadBalancer.Spec.Ports[1].Port).To(BeEquivalentTo(29092))
-	Expect(loadBalancer.Spec.Ports[1].TargetPort.IntVal).To(BeEquivalentTo(29092))
+	Expect(loadBalancer.Spec.Ports).To(HaveLen(4))
+	for i, port := range loadBalancer.Spec.Ports {
+		if i == 3 {
+			break
+		}
+		Expect(port.Name).To(Equal(fmt.Sprintf("broker-%d", i)))
+		Expect(port.Protocol).To(Equal(corev1.ProtocolTCP))
+		Expect(port.Port).To(BeEquivalentTo(19090 + i))
+		Expect(port.TargetPort.IntVal).To(BeEquivalentTo(19090 + i))
+	}
+	Expect(loadBalancer.Spec.Ports[3].Name).To(Equal("tcp-all-broker"))
+	Expect(loadBalancer.Spec.Ports[3].Protocol).To(Equal(corev1.ProtocolTCP))
+	Expect(loadBalancer.Spec.Ports[3].Port).To(BeEquivalentTo(29092))
+	Expect(loadBalancer.Spec.Ports[3].TargetPort.IntVal).To(BeEquivalentTo(29092))
 
 	var configMap corev1.ConfigMap
 	configMapName := fmt.Sprintf("envoy-config-test-%s", kafkaCluster.Name)
@@ -68,6 +72,7 @@ func expectEnvoy(kafkaCluster *v1beta1.KafkaCluster) {
 
 	expectEnvoyIngressLabels(configMap.Labels, "test", kafkaCluster.Name)
 	Expect(configMap.Data).To(HaveKey("envoy.yaml"))
+	svcTemplate := fmt.Sprintf("%s-%s.%s.svc.%s", kafkaCluster.Name, "%s", kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain())
 	Expect(configMap.Data["envoy.yaml"]).To(Equal(fmt.Sprintf(`admin:
   accessLogPath: /tmp/admin_access.log
   address:
@@ -79,16 +84,32 @@ staticResources:
   - connectTimeout: 1s
     hosts:
     - socketAddress:
-        address: %s-0.%s.svc.%s
-        portValue: 9733
+        address: %s
+        portValue: 9094
     http2ProtocolOptions: {}
     name: broker-0
     type: STRICT_DNS
   - connectTimeout: 1s
     hosts:
     - socketAddress:
-        address: %s-all-broker.%s.svc.%s
-        portValue: 9733
+        address: %s
+        portValue: 9094
+    http2ProtocolOptions: {}
+    name: broker-1
+    type: STRICT_DNS
+  - connectTimeout: 1s
+    hosts:
+    - socketAddress:
+        address: %s
+        portValue: 9094
+    http2ProtocolOptions: {}
+    name: broker-2
+    type: STRICT_DNS
+  - connectTimeout: 1s
+    hosts:
+    - socketAddress:
+        address: %s
+        portValue: 9094
     http2ProtocolOptions: {}
     name: all-brokers
     type: STRICT_DNS
@@ -96,13 +117,33 @@ staticResources:
   - address:
       socketAddress:
         address: 0.0.0.0
-        portValue: 11202
+        portValue: 19090
     filterChains:
     - filters:
       - config:
           cluster: broker-0
           stat_prefix: broker_tcp-0
         name: envoy.filters.network.tcp_proxy
+  - address:
+      socketAddress:
+        address: 0.0.0.0
+        portValue: 19091
+    filterChains:
+    - filters:
+      - config:
+          cluster: broker-1
+          stat_prefix: broker_tcp-1
+        name: envoy.filters.network.tcp_proxy
+  - address:
+      socketAddress:
+        address: 0.0.0.0
+        portValue: 19092
+    filterChains:
+    - filters:
+      - config:
+          cluster: broker-2
+          stat_prefix: broker_tcp-2
+        name: envoy.filters.network.tcp_proxy
   - address:
       socketAddress:
         address: 0.0.0.0
@@ -113,8 +154,7 @@ staticResources:
           cluster: all-brokers
           stat_prefix: all-brokers
         name: envoy.filters.network.tcp_proxy
-`, kafkaCluster.Name, kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(),
-		kafkaCluster.Name, kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain())))
+`, fmt.Sprintf(svcTemplate, "0"), fmt.Sprintf(svcTemplate, "1"), fmt.Sprintf(svcTemplate, "2"), fmt.Sprintf(svcTemplate, "all-broker"))))
 
 	var deployment appsv1.Deployment
 	deploymentName := fmt.Sprintf("envoy-test-%s", kafkaCluster.Name)
@@ -138,7 +178,17 @@ staticResources:
 	Expect(container.Ports).To(ConsistOf(
 		corev1.ContainerPort{
 			Name:          "broker-0",
-			ContainerPort: 11202,
+			ContainerPort: 19090,
+			Protocol:      corev1.ProtocolTCP,
+		},
+		corev1.ContainerPort{
+			Name:          "broker-1",
+			ContainerPort: 19091,
+			Protocol:      corev1.ProtocolTCP,
+		},
+		corev1.ContainerPort{
+			Name:          "broker-2",
+			ContainerPort: 19092,
 			Protocol:      corev1.ProtocolTCP,
 		},
 		corev1.ContainerPort{
diff --git a/controllers/tests/kafkacluster_controller_externalnodeport_test.go b/controllers/tests/kafkacluster_controller_externalnodeport_test.go
index 7869369c..e3370ad6 100644
--- a/controllers/tests/kafkacluster_controller_externalnodeport_test.go
+++ b/controllers/tests/kafkacluster_controller_externalnodeport_test.go
@@ -23,6 +23,7 @@ import (
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -30,6 +31,24 @@ import (
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
 
+var (
+	defaultStorageConfig = []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"),
+					},
+				},
+			},
+		},
+	}
+)
+
 var _ = Describe("KafkaClusterNodeportExternalAccess", func() {
 	var (
 		count              uint64 = 0
@@ -53,18 +72,6 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", func() {
 		kafkaClusterCRName = fmt.Sprintf("kafkacluster-%v", count)
 		kafkaCluster = createMinimalKafkaClusterCR(kafkaClusterCRName, namespace)
 
-		// update the external listener config with a nodeport listener
-		kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
-			{
-				CommonListenerSpec: v1beta1.CommonListenerSpec{
-					Name:          "test",
-					ContainerPort: 9733,
-				},
-				ExternalStartingPort: 31123,
-				HostnameOverride:     "test-host",
-				AccessMethod:         corev1.ServiceTypeNodePort,
-			},
-		}
 	})
 
 	JustBeforeEach(func() {
@@ -86,39 +93,198 @@ var _ = Describe("KafkaClusterNodeportExternalAccess", func() {
 		kafkaCluster = nil
 	})
 
-	It("reconciles the service successfully", func() {
-		var svc corev1.Service
-		svcName := fmt.Sprintf("%s-0-test", kafkaClusterCRName)
-		Eventually(func() error {
-			err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: svcName}, &svc)
-			return err
-		}).Should(Succeed())
-
-		Expect(svc.Labels).To(Equal(map[string]string{
-			"app":      "kafka",
-			"brokerId": "0",
-			"kafka_cr": kafkaCluster.Name,
-		}))
-
-		Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeNodePort))
-		Expect(svc.Spec.Selector).To(Equal(map[string]string{
-			"app":      "kafka",
-			"brokerId": "0",
-			"kafka_cr": kafkaCluster.Name,
-		}))
-
-		Expect(svc.Spec.Ports).To(HaveLen(1))
-		Expect(svc.Spec.Ports[0].Name).To(Equal("broker-0"))
-		Expect(svc.Spec.Ports[0].Protocol).To(Equal(corev1.ProtocolTCP))
-		Expect(svc.Spec.Ports[0].Port).To(BeEquivalentTo(9733))
-		Expect(svc.Spec.Ports[0].TargetPort.IntVal).To(BeEquivalentTo(9733))
-
-		Expect(svc.Spec.Ports).To(ConsistOf(corev1.ServicePort{
-			Name:       "broker-0",
-			Protocol:   "TCP",
-			Port:       9733,
-			TargetPort: intstr.FromInt(9733),
-			NodePort:   31123,
-		}))
+	When("NodePortExternalIP is configured", func() {
+		BeforeEach(func() {
+			// update the external listener config with a nodeport listener
+			kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
+				{
+					CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Name:          "test",
+						ContainerPort: 9733,
+					},
+					ExternalStartingPort: 31123,
+					AccessMethod:         corev1.ServiceTypeNodePort,
+				},
+			}
+			kafkaCluster.Spec.BrokerConfigGroups = map[string]v1beta1.BrokerConfig{
+				"br-0": {
+					StorageConfigs: defaultStorageConfig,
+					NodePortExternalIP: map[string]string{
+						"test": "1.2.3.4",
+					},
+				},
+				"br-1": {
+					StorageConfigs: defaultStorageConfig,
+					NodePortExternalIP: map[string]string{
+						"test": "1.2.3.5",
+					},
+				},
+				"br-2": {
+					StorageConfigs: defaultStorageConfig,
+					NodePortExternalIP: map[string]string{
+						"test": "1.2.3.6",
+					},
+				},
+			}
+			kafkaCluster.Spec.Brokers = []v1beta1.Broker{
+				{
+					Id:                0,
+					BrokerConfigGroup: "br-0",
+				},
+				{
+					Id:                1,
+					BrokerConfigGroup: "br-1",
+				},
+				{
+					Id:                2,
+					BrokerConfigGroup: "br-2",
+				},
+			}
+		})
+
+		It("reconciles the service successfully", func() {
+			var svc corev1.Service
+			svcName := fmt.Sprintf("%s-0-test", kafkaClusterCRName)
+			Eventually(func() error {
+				err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: svcName}, &svc)
+				return err
+			}).Should(Succeed())
+
+			Expect(svc.Labels).To(Equal(map[string]string{
+				"app":      "kafka",
+				"brokerId": "0",
+				"kafka_cr": kafkaCluster.Name,
+			}))
+
+			Expect(svc.Spec.Type).To(Equal(corev1.ServiceTypeNodePort))
+			Expect(svc.Spec.Selector).To(Equal(map[string]string{
+				"app":      "kafka",
+				"brokerId": "0",
+				"kafka_cr": kafkaCluster.Name,
+			}))
+
+			Expect(svc.Spec.Ports).To(HaveLen(1))
+			Expect(svc.Spec.Ports[0].Name).To(Equal("broker-0"))
+			Expect(svc.Spec.Ports[0].Protocol).To(Equal(corev1.ProtocolTCP))
+			Expect(svc.Spec.Ports[0].Port).To(BeEquivalentTo(9733))
+			Expect(svc.Spec.Ports[0].TargetPort.IntVal).To(BeEquivalentTo(9733))
+
+			Expect(svc.Spec.Ports).To(ConsistOf(corev1.ServicePort{
+				Name:       "broker-0",
+				Protocol:   "TCP",
+				Port:       9733,
+				TargetPort: intstr.FromInt(9733),
+				NodePort:   31123,
+			}))
+
+			// check status
+			err := k8sClient.Get(context.TODO(), types.NamespacedName{
+				Name:      kafkaCluster.Name,
+				Namespace: kafkaCluster.Namespace,
+			}, kafkaCluster)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
+				InternalListeners: map[string]v1beta1.ListenerStatusList{
+					"internal": {
+						{
+							Name:    "any-broker",
+							Address: fmt.Sprintf("%s-all-broker.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-0",
+							Address: fmt.Sprintf("%s-0.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-1",
+							Address: fmt.Sprintf("%s-1.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-2",
+							Address: fmt.Sprintf("%s-2.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+					},
+				},
+				ExternalListeners: map[string]v1beta1.ListenerStatusList{
+					"test": {
+						{
+							Name:    "broker-0",
+							Address: "1.2.3.4:9733",
+						},
+						{
+							Name:    "broker-1",
+							Address: "1.2.3.5:9733",
+						},
+						{
+							Name:    "broker-2",
+							Address: "1.2.3.6:9733",
+						},
+					},
+				},
+			}))
+		})
+	})
+
+	When("hostnameOverride is configured", func() {
+		BeforeEach(func() {
+			kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
+				{
+					CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Name:          "test",
+						ContainerPort: 9733,
+					},
+					ExternalStartingPort: 30100,
+					HostnameOverride:     ".external.nodeport.com",
+					AccessMethod:         corev1.ServiceTypeNodePort,
+				},
+			}
+		})
+
+		It("reconciles the status successfully", func() {
+			err := k8sClient.Get(context.TODO(), types.NamespacedName{
+				Name:      kafkaCluster.Name,
+				Namespace: kafkaCluster.Namespace,
+			}, kafkaCluster)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
+				InternalListeners: map[string]v1beta1.ListenerStatusList{
+					"internal": {
+						{
+							Name:    "any-broker",
+							Address: fmt.Sprintf("%s-all-broker.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-0",
+							Address: fmt.Sprintf("%s-0.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-1",
+							Address: fmt.Sprintf("%s-1.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-2",
+							Address: fmt.Sprintf("%s-2.kafka-nodeport-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+					},
+				},
+				ExternalListeners: map[string]v1beta1.ListenerStatusList{
+					"test": {
+						{
+							Name:    "broker-0",
+							Address: fmt.Sprintf("%s-0-test.kafka-nodeport-%d.external.nodeport.com:30100", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-1",
+							Address: fmt.Sprintf("%s-1-test.kafka-nodeport-%d.external.nodeport.com:30101", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-2",
+							Address: fmt.Sprintf("%s-2-test.kafka-nodeport-%d.external.nodeport.com:30102", kafkaCluster.Name, count),
+						},
+					},
+				},
+			}))
+		})
 	})
 })
diff --git a/controllers/tests/kafkacluster_controller_istioingress_test.go b/controllers/tests/kafkacluster_controller_istioingress_test.go
index 5979a053..2ed9a853 100644
--- a/controllers/tests/kafkacluster_controller_istioingress_test.go
+++ b/controllers/tests/kafkacluster_controller_istioingress_test.go
@@ -64,6 +64,18 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 
 		kafkaClusterCRName = fmt.Sprintf("kafkacluster-%v", count)
 		kafkaCluster = createMinimalKafkaClusterCR(kafkaClusterCRName, namespace)
+
+		kafkaCluster.Spec.IngressController = istioingress.IngressControllerName
+		kafkaCluster.Spec.ListenersConfig.ExternalListeners = []v1beta1.ExternalListenerConfig{
+			{
+				CommonListenerSpec: v1beta1.CommonListenerSpec{
+					Type:          "plaintext",
+					Name:          "external",
+					ContainerPort: 9094,
+				},
+				ExternalStartingPort: 19090,
+			},
+		}
 	})
 
 	JustBeforeEach(func() {
@@ -75,6 +87,37 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 		err = k8sClient.Create(context.TODO(), kafkaCluster)
 		Expect(err).NotTo(HaveOccurred())
 
+		svcName := fmt.Sprintf("meshgateway-external-%s", kafkaCluster.Name)
+		svcFromMeshGateway := corev1.Service{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      svcName,
+				Namespace: namespace,
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					// other ports omitted
+					{
+						Name:     "tcp-all-broker",
+						Port:     29092, // from MeshGateway (guarded by the tests)
+						Protocol: corev1.ProtocolTCP,
+					},
+				},
+			},
+			Status: corev1.ServiceStatus{
+				LoadBalancer: corev1.LoadBalancerStatus{
+					Ingress: []corev1.LoadBalancerIngress{
+						{
+							Hostname: "ingress.test.host.com",
+						},
+					},
+				},
+			},
+		}
+		err = k8sClient.Create(context.TODO(), &svcFromMeshGateway)
+		Expect(err).NotTo(HaveOccurred())
+		err = k8sClient.Status().Update(context.TODO(), &svcFromMeshGateway)
+		Expect(err).NotTo(HaveOccurred())
+
 		waitForClusterRunningState(kafkaCluster, namespace)
 	})
 
@@ -92,14 +135,14 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 
 		It("creates Istio ingress related objects", func() {
 			var meshGateway istioOperatorApi.MeshGateway
-			meshGatewayName := fmt.Sprintf("meshgateway-test-%s", kafkaCluster.Name)
+			meshGatewayName := fmt.Sprintf("meshgateway-external-%s", kafkaCluster.Name)
 			Eventually(func() error {
 				err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: meshGatewayName}, &meshGateway)
 				return err
 			}).Should(Succeed())
 
 			meshGatewayConf := meshGateway.Spec.MeshGatewayConfiguration
-			ExpectIstioIngressLabels(meshGatewayConf.Labels, "test", kafkaClusterCRName)
+			ExpectIstioIngressLabels(meshGatewayConf.Labels, "external", kafkaClusterCRName)
 			Expect(meshGatewayConf.ServiceType).To(Equal(corev1.ServiceTypeLoadBalancer))
 			baseConf := meshGatewayConf.BaseK8sResourceConfigurationWithHPAWithoutImage
 			Expect(baseConf.ReplicaCount).To(Equal(util.Int32Pointer(1)))
@@ -125,8 +168,18 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 			Expect(meshGateway.Spec.Ports).To(ConsistOf(
 				corev1.ServicePort{
 					Name:       "tcp-broker-0",
-					Port:       11202,
-					TargetPort: intstr.FromInt(11202),
+					Port:       19090,
+					TargetPort: intstr.FromInt(19090),
+				},
+				corev1.ServicePort{
+					Name:       "tcp-broker-1",
+					Port:       19091,
+					TargetPort: intstr.FromInt(19091),
+				},
+				corev1.ServicePort{
+					Name:       "tcp-broker-2",
+					Port:       19092,
+					TargetPort: intstr.FromInt(19092),
 				},
 				corev1.ServicePort{
 					Name:       "tcp-all-broker",
@@ -136,22 +189,36 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 			Expect(meshGateway.Spec.Type).To(Equal(istioOperatorApi.GatewayTypeIngress))
 
 			var gateway v1alpha3.Gateway
-			gatewayName := fmt.Sprintf("%s-test-gateway", kafkaCluster.Name)
+			gatewayName := fmt.Sprintf("%s-external-gateway", kafkaCluster.Name)
 			Eventually(func() error {
 				err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: gatewayName}, &gateway)
 				return err
 			}).Should(Succeed())
 
-			ExpectIstioIngressLabels(gateway.Labels, "test", kafkaClusterCRName)
-			ExpectIstioIngressLabels(gateway.Spec.Selector, "test", kafkaClusterCRName)
+			ExpectIstioIngressLabels(gateway.Labels, "external", kafkaClusterCRName)
+			ExpectIstioIngressLabels(gateway.Spec.Selector, "external", kafkaClusterCRName)
 			Expect(gateway.Spec.Servers).To(ConsistOf(
 				v1alpha3.Server{
 					Port: &v1alpha3.Port{
-						Number:   11202,
+						Number:   19090,
 						Protocol: "TCP",
 						Name:     "tcp-broker-0"},
 					Hosts: []string{"*"},
 				},
+				v1alpha3.Server{
+					Port: &v1alpha3.Port{
+						Number:   19091,
+						Protocol: "TCP",
+						Name:     "tcp-broker-1"},
+					Hosts: []string{"*"},
+				},
+				v1alpha3.Server{
+					Port: &v1alpha3.Port{
+						Number:   19092,
+						Protocol: "TCP",
+						Name:     "tcp-broker-2"},
+					Hosts: []string{"*"},
+				},
 				v1alpha3.Server{
 					Port: &v1alpha3.Port{
 						Number:   29092,
@@ -162,23 +229,41 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 				}))
 
 			var virtualService v1alpha3.VirtualService
-			virtualServiceName := fmt.Sprintf("%s-test-virtualservice", kafkaCluster.Name)
+			virtualServiceName := fmt.Sprintf("%s-external-virtualservice", kafkaCluster.Name)
 			Eventually(func() error {
 				err := k8sClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: virtualServiceName}, &virtualService)
 				return err
 			}).Should(Succeed())
 
-			ExpectIstioIngressLabels(virtualService.Labels, "test", kafkaClusterCRName)
+			ExpectIstioIngressLabels(virtualService.Labels, "external", kafkaClusterCRName)
 			Expect(virtualService.Spec).To(Equal(v1alpha3.VirtualServiceSpec{
 				Hosts:    []string{"*"},
-				Gateways: []string{fmt.Sprintf("%s-test-gateway", kafkaClusterCRName)},
+				Gateways: []string{fmt.Sprintf("%s-external-gateway", kafkaClusterCRName)},
 				TCP: []v1alpha3.TCPRoute{
 					{
-						Match: []v1alpha3.L4MatchAttributes{{Port: util.IntPointer(11202)}},
+						Match: []v1alpha3.L4MatchAttributes{{Port: util.IntPointer(19090)}},
 						Route: []*v1alpha3.RouteDestination{{
 							Destination: &v1alpha3.Destination{
 								Host: "kafkacluster-1-0",
-								Port: &v1alpha3.PortSelector{Number: 9733},
+								Port: &v1alpha3.PortSelector{Number: 9094},
+							},
+						}},
+					},
+					{
+						Match: []v1alpha3.L4MatchAttributes{{Port: util.IntPointer(19091)}},
+						Route: []*v1alpha3.RouteDestination{{
+							Destination: &v1alpha3.Destination{
+								Host: "kafkacluster-1-1",
+								Port: &v1alpha3.PortSelector{Number: 9094},
+							},
+						}},
+					},
+					{
+						Match: []v1alpha3.L4MatchAttributes{{Port: util.IntPointer(19092)}},
+						Route: []*v1alpha3.RouteDestination{{
+							Destination: &v1alpha3.Destination{
+								Host: "kafkacluster-1-2",
+								Port: &v1alpha3.PortSelector{Number: 9094},
 							},
 						}},
 					},
@@ -187,12 +272,119 @@ var _ = Describe("KafkaClusterIstioIngressController", func() {
 						Route: []*v1alpha3.RouteDestination{{
 							Destination: &v1alpha3.Destination{
 								Host: "kafkacluster-1-all-broker",
-								Port: &v1alpha3.PortSelector{Number: 9733},
+								Port: &v1alpha3.PortSelector{Number: 9094},
 							},
 						}},
 					},
 				},
 			}))
+
+			// expect kafkaCluster listener status
+			err = k8sClient.Get(context.TODO(), types.NamespacedName{
+				Name:      kafkaCluster.Name,
+				Namespace: kafkaCluster.Namespace,
+			}, kafkaCluster)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
+				InternalListeners: map[string]v1beta1.ListenerStatusList{
+					"internal": {
+						{
+							Name:    "any-broker",
+							Address: fmt.Sprintf("%s-all-broker.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-0",
+							Address: fmt.Sprintf("%s-0.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-1",
+							Address: fmt.Sprintf("%s-1.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-2",
+							Address: fmt.Sprintf("%s-2.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+					},
+				},
+				ExternalListeners: map[string]v1beta1.ListenerStatusList{
+					"external": {
+						{
+							Name:    "any-broker",
+							Address: "ingress.test.host.com:29092",
+						},
+						{
+							Name:    "broker-0",
+							Address: "ingress.test.host.com:19090",
+						},
+						{
+							Name:    "broker-1",
+							Address: "ingress.test.host.com:19091",
+						},
+						{
+							Name:    "broker-2",
+							Address: "ingress.test.host.com:19092",
+						},
+					},
+				},
+			}))
+		})
+	})
+
+	When("Headless mode is turned on", func() {
+		BeforeEach(func() {
+			kafkaCluster.Spec.HeadlessServiceEnabled = true
+		})
+
+		It("does not add the all-broker service to the listener status", func() {
+			err := k8sClient.Get(context.TODO(), types.NamespacedName{
+				Name:      kafkaCluster.Name,
+				Namespace: kafkaCluster.Namespace,
+			}, kafkaCluster)
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
+				InternalListeners: map[string]v1beta1.ListenerStatusList{
+					"internal": {
+						{
+							Name:    "headless",
+							Address: fmt.Sprintf("%s-headless.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-0",
+							Address: fmt.Sprintf("%s-0.%s-headless.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-1",
+							Address: fmt.Sprintf("%s-1.%s-headless.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, kafkaCluster.Name, count),
+						},
+						{
+							Name:    "broker-2",
+							Address: fmt.Sprintf("%s-2.%s-headless.kafka-istioingress-%d.svc.cluster.local:29092", kafkaCluster.Name, kafkaCluster.Name, count),
+						},
+					},
+				},
+				ExternalListeners: map[string]v1beta1.ListenerStatusList{
+					"external": {
+						{
+							Name:    "any-broker",
+							Address: "ingress.test.host.com:29092",
+						},
+						{
+							Name:    "broker-0",
+							Address: "ingress.test.host.com:19090",
+						},
+						{
+							Name:    "broker-1",
+							Address: "ingress.test.host.com:19091",
+						},
+						{
+							Name:    "broker-2",
+							Address: "ingress.test.host.com:19092",
+						},
+					},
+				},
+			}))
 		})
 	})
 })
diff --git a/controllers/tests/kafkacluster_controller_kafka_test.go b/controllers/tests/kafkacluster_controller_kafka_test.go
index 1eaaae4e..b22e3de3 100644
--- a/controllers/tests/kafkacluster_controller_kafka_test.go
+++ b/controllers/tests/kafkacluster_controller_kafka_test.go
@@ -44,6 +44,8 @@ func expectKafka(kafkaCluster *v1beta1.KafkaCluster) {
 		expectKafkaBrokerPod(kafkaCluster, broker)
 	}
 
+	expectKafkaCRStatus(kafkaCluster)
+
 	// TODO test reconcile PKI?
 	// TODO test reconcileKafkaPodDelete
 }
@@ -80,14 +82,15 @@ func expectKafkaAllBrokerService(kafkaCluster *v1beta1.KafkaCluster) {
 		corev1.ServicePort{
 			Name:       "tcp-test",
 			Protocol:   "TCP",
-			Port:       9733,
-			TargetPort: intstr.FromInt(9733),
+			Port:       9094,
+			TargetPort: intstr.FromInt(9094),
 		}))
 }
 
 func expectKafkaPDB(kafkaCluster *v1beta1.KafkaCluster) {
 	// get current CR
 	err := k8sClient.Get(context.TODO(), types.NamespacedName{Name: kafkaCluster.Name, Namespace: kafkaCluster.Namespace}, kafkaCluster)
+	Expect(err).NotTo(HaveOccurred())
 
 	// set PDB and reset status
 	kafkaCluster.Spec.DisruptionBudget = v1beta1.DisruptionBudget{
@@ -115,7 +118,7 @@ func expectKafkaPDB(kafkaCluster *v1beta1.KafkaCluster) {
 	// make assertions
 	Expect(pdb.Labels).To(HaveKeyWithValue("app", "kafka"))
 	Expect(pdb.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
-	Expect(pdb.Spec.MinAvailable).To(Equal(util.IntstrPointer(1)))
+	Expect(pdb.Spec.MinAvailable).To(Equal(util.IntstrPointer(3)))
 	Expect(pdb.Spec.Selector).NotTo(BeNil())
 	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue("app", "kafka"))
 	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
@@ -130,19 +133,20 @@ func expectKafkaPVC(kafkaCluster *v1beta1.KafkaCluster) {
 			client.ListOption(client.MatchingLabels(map[string]string{"app": "kafka", "kafka_cr": kafkaCluster.Name})))
 	}).Should(Succeed())
 
-	Expect(pvcs.Items).To(HaveLen(1))
-	pvc := pvcs.Items[0]
-	Expect(pvc.GenerateName).To(Equal(fmt.Sprintf("%s-0-storage-0-", kafkaCluster.Name)))
-	Expect(pvc.Labels).To(HaveKeyWithValue("app", "kafka"))
-	Expect(pvc.Labels).To(HaveKeyWithValue("brokerId", "0"))
-	Expect(pvc.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
-	Expect(pvc.Annotations).To(HaveKeyWithValue("mountPath", "/kafka-logs"))
-	Expect(pvc.Spec.AccessModes).To(ConsistOf(corev1.ReadWriteOnce))
-	Expect(pvc.Spec.Resources).To(Equal(corev1.ResourceRequirements{
-		Requests: corev1.ResourceList{
-			"storage": resource.MustParse("10Gi"),
-		},
-	}))
+	Expect(pvcs.Items).To(HaveLen(3))
+	for i, pvc := range pvcs.Items {
+		Expect(pvc.GenerateName).To(Equal(fmt.Sprintf("%s-%d-storage-0-", kafkaCluster.Name, i)))
+		Expect(pvc.Labels).To(HaveKeyWithValue("app", "kafka"))
+		Expect(pvc.Labels).To(HaveKeyWithValue("brokerId", strconv.Itoa(i)))
+		Expect(pvc.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
+		Expect(pvc.Annotations).To(HaveKeyWithValue("mountPath", "/kafka-logs"))
+		Expect(pvc.Spec.AccessModes).To(ConsistOf(corev1.ReadWriteOnce))
+		Expect(pvc.Spec.Resources).To(Equal(corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				"storage": resource.MustParse("10Gi"),
+			},
+		}))
+	}
 }
 
 func expectKafkaBrokerConfigmap(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Broker) {
@@ -158,17 +162,17 @@ func expectKafkaBrokerConfigmap(kafkaCluster *v1beta1.KafkaCluster, broker v1bet
 	Expect(configMap.Labels).To(HaveKeyWithValue("kafka_cr", kafkaCluster.Name))
 	Expect(configMap.Labels).To(HaveKeyWithValue("brokerId", strconv.Itoa(int(broker.Id))))
 
-	Expect(configMap.Data).To(HaveKeyWithValue("broker-config", fmt.Sprintf(`advertised.listeners=TEST://test-host:11202,INTERNAL://kafkacluster-1-0.kafka-1.svc.cluster.local:29092,CONTROLLER://kafkacluster-1-0.kafka-1.svc.cluster.local:29093
+	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-0.kafka-1.svc.cluster.local:29092
+cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafkacluster-1-%d.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:
-listeners=INTERNAL://:29092,CONTROLLER://:29093,TEST://:9733
+listeners=INTERNAL://:29092,CONTROLLER://:29093,TEST://:9094
 log.dirs=/kafka-logs/kafka
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
-zookeeper.connect=/`, broker.Id)))
+zookeeper.connect=/`, 19090+broker.Id, broker.Id, broker.Id, broker.Id, broker.Id)))
 
 	// assert log4j?
 }
@@ -202,8 +206,8 @@ func expectKafkaBrokerService(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1
 		corev1.ServicePort{
 			Name:       "tcp-test",
 			Protocol:   "TCP",
-			Port:       9733,
-			TargetPort: intstr.FromInt(9733),
+			Port:       9094,
+			TargetPort: intstr.FromInt(9094),
 		},
 		corev1.ServicePort{
 			Name:       "metrics",
@@ -223,7 +227,7 @@ func expectKafkaBrokerPod(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Bro
 	Eventually(func() ([]corev1.Pod, error) {
 		err := k8sClient.List(context.Background(), &podList,
 			client.ListOption(client.InNamespace(kafkaCluster.Namespace)),
-			client.ListOption(client.MatchingLabels(map[string]string{"app": "kafka", "kafka_cr": kafkaCluster.Name})))
+			client.ListOption(client.MatchingLabels(map[string]string{"app": "kafka", "kafka_cr": kafkaCluster.Name, "brokerId": strconv.Itoa(int(broker.Id))})))
 		return podList.Items, err
 	}).Should(HaveLen(1))
 
@@ -308,7 +312,7 @@ func expectKafkaBrokerPod(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Bro
 		},
 
 		// the name of the PVC is dynamically created - no exact match
-		WithTransform(getVolumeName, Equal(fmt.Sprintf("kafka-data-%d", broker.Id))),
+		WithTransform(getVolumeName, Equal("kafka-data-0")),
 	))
 
 	Expect(pod.Spec.RestartPolicy).To(Equal(corev1.RestartPolicyNever))
@@ -316,3 +320,57 @@ func expectKafkaBrokerPod(kafkaCluster *v1beta1.KafkaCluster, broker v1beta1.Bro
 
 	// expect some other fields
 }
+
+func expectKafkaCRStatus(kafkaCluster *v1beta1.KafkaCluster) {
+	err := k8sClient.Get(context.TODO(), types.NamespacedName{
+		Name:      kafkaCluster.Name,
+		Namespace: kafkaCluster.Namespace,
+	}, kafkaCluster)
+	Expect(err).NotTo(HaveOccurred())
+
+	Expect(kafkaCluster.Status.State).To(Equal(v1beta1.KafkaClusterRunning))
+	Expect(kafkaCluster.Status.AlertCount).To(Equal(0))
+
+	Expect(kafkaCluster.Status.ListenerStatuses).To(Equal(v1beta1.ListenerStatuses{
+		InternalListeners: map[string]v1beta1.ListenerStatusList{
+			"internal": {
+				{
+					Name:    "any-broker",
+					Address: fmt.Sprintf("%s-all-broker.kafka-1.svc.cluster.local:29092", kafkaCluster.Name),
+				},
+				{
+					Name:    "broker-0",
+					Address: fmt.Sprintf("%s-0.kafka-1.svc.cluster.local:29092", kafkaCluster.Name),
+				},
+				{
+					Name:    "broker-1",
+					Address: fmt.Sprintf("%s-1.kafka-1.svc.cluster.local:29092", kafkaCluster.Name),
+				},
+				{
+					Name:    "broker-2",
+					Address: fmt.Sprintf("%s-2.kafka-1.svc.cluster.local:29092", kafkaCluster.Name),
+				},
+			},
+		},
+		ExternalListeners: map[string]v1beta1.ListenerStatusList{
+			"test": {
+				{
+					Name:    "any-broker",
+					Address: "test.host.com:29092",
+				},
+				{
+					Name:    "broker-0",
+					Address: "test.host.com:19090",
+				},
+				{
+					Name:    "broker-1",
+					Address: "test.host.com:19091",
+				},
+				{
+					Name:    "broker-2",
+					Address: "test.host.com:19092",
+				},
+			},
+		},
+	}))
+}
diff --git a/controllers/tests/kafkacluster_controller_test.go b/controllers/tests/kafkacluster_controller_test.go
index 192338e3..3e00abf0 100644
--- a/controllers/tests/kafkacluster_controller_test.go
+++ b/controllers/tests/kafkacluster_controller_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"sync/atomic"
+	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -26,6 +27,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/runtime/log"
 
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
@@ -49,6 +51,7 @@ var _ = Describe("KafkaCluster", func() {
 		}
 
 		kafkaCluster = createMinimalKafkaClusterCR(fmt.Sprintf("kafkacluster-%d", count), namespace)
+		kafkaCluster.Spec.ListenersConfig.ExternalListeners[0].HostnameOverride = ""
 		kafkaCluster.Spec.CruiseControlConfig = v1beta1.CruiseControlConfig{
 			TopicConfig: &v1beta1.TopicConfig{
 				Partitions:        7,
@@ -68,6 +71,23 @@ var _ = Describe("KafkaCluster", func() {
 		err = k8sClient.Create(context.TODO(), kafkaCluster)
 		Expect(err).NotTo(HaveOccurred())
 
+		// assign host to envoy LB
+		envoyLBService := &corev1.Service{}
+		Eventually(func() error {
+			return k8sClient.Get(context.TODO(), types.NamespacedName{
+				Name:      fmt.Sprintf("envoy-loadbalancer-test-%s", kafkaCluster.Name),
+				Namespace: namespace,
+			}, envoyLBService)
+		}, 5*time.Second, 100*time.Millisecond).Should(Succeed())
+
+		envoyLBService.Status.LoadBalancer.Ingress = []corev1.LoadBalancerIngress{{
+			Hostname: "test.host.com",
+		}}
+
+		log.Log.V(-1).Info("envoy service updated", "spec", envoyLBService)
+		err = k8sClient.Status().Update(context.TODO(), envoyLBService)
+		Expect(err).NotTo(HaveOccurred())
+
 		waitForClusterRunningState(kafkaCluster, namespace)
 	})
 
diff --git a/controllers/tests/suite_test.go b/controllers/tests/suite_test.go
index 156a1ce4..634fb8d9 100644
--- a/controllers/tests/suite_test.go
+++ b/controllers/tests/suite_test.go
@@ -135,6 +135,14 @@ var _ = BeforeSuite(func(done Done) {
 
 	scale.MockNewCruiseControlScaler()
 
+	mockKafkaClients = make(map[types.NamespacedName]kafkaclient.KafkaClient)
+
+	// mock the creation of Kafka clients
+	controllers.SetNewKafkaFromCluster(
+		func(k8sclient client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (kafkaclient.KafkaClient, error) {
+			return getMockedKafkaClientForCluster(cluster), nil
+		})
+
 	kafkaClusterReconciler := controllers.KafkaClusterReconciler{
 		Client:              mgr.GetClient(),
 		DirectClient:        mgr.GetAPIReader(),
@@ -146,14 +154,6 @@ var _ = BeforeSuite(func(done Done) {
 	err = controllers.SetupKafkaClusterWithManager(mgr, kafkaClusterReconciler.Log).Complete(&kafkaClusterReconciler)
 	Expect(err).NotTo(HaveOccurred())
 
-	mockKafkaClients = make(map[types.NamespacedName]kafkaclient.KafkaClient)
-
-	// mock the creation of Kafka clients
-	controllers.SetNewKafkaFromCluster(
-		func(k8sclient client.Client, cluster *banzaicloudv1beta1.KafkaCluster) (kafkaclient.KafkaClient, error) {
-			return getMockedKafkaClientForCluster(cluster), nil
-		})
-
 	err = controllers.SetupKafkaTopicWithManager(mgr)
 	Expect(err).NotTo(HaveOccurred())
 
diff --git a/pkg/kafkaclient/config_test.go b/pkg/kafkaclient/config_test.go
index d3f0f466..42d87f85 100644
--- a/pkg/kafkaclient/config_test.go
+++ b/pkg/kafkaclient/config_test.go
@@ -17,9 +17,10 @@ package kafkaclient
 import (
 	"testing"
 
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/pki"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 type mockClient struct {
@@ -42,23 +43,6 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	return cluster
 }
 
-func TestGenerateKafkaAddress(t *testing.T) {
-	cluster := newMockCluster()
-	cluster.Spec.HeadlessServiceEnabled = true
-	generatedHeadless := generateKafkaAddress(cluster)
-	expected := "test-headless.test.svc.cluster.local:80"
-	if generatedHeadless != expected {
-		t.Error("Expected kafka address:", expected, "Got:", generatedHeadless)
-	}
-
-	cluster.Spec.HeadlessServiceEnabled = false
-	generatedAllBroker := generateKafkaAddress(cluster)
-	expected = "test-all-broker.test.svc.cluster.local:80"
-	if generatedAllBroker != expected {
-		t.Error("Expected kafka address:", expected, "Got:", generatedAllBroker)
-	}
-}
-
 func TestClusterConfig(t *testing.T) {
 	cluster := newMockCluster()
 	_, err := ClusterConfig(&mockClient{}, cluster)
diff --git a/pkg/pki/certmanagerpki/certmanager_pki_test.go b/pkg/pki/certmanagerpki/certmanager_pki_test.go
index bc277543..c8de3983 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki_test.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki_test.go
@@ -25,6 +25,7 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
 	certutil "github.com/banzaicloud/kafka-operator/pkg/util/cert"
 	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
@@ -97,33 +98,33 @@ func TestReconcilePKI(t *testing.T) {
 	ctx := context.Background()
 
 	manager.client.Create(ctx, newServerSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 			t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 		}
 	}
 
 	manager.client.Create(ctx, newControllerSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 			t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 		}
 	}
 
 	manager.client.Create(ctx, newCASecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		t.Error("Expected successful reconcile, got:", err)
 	}
 
 	cluster.Spec.ListenersConfig.SSLSecrets.Create = false
 	manager = newMock(cluster)
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err == nil {
 		t.Error("Expected error got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected not ready error, got:", reflect.TypeOf(err))
 	}
 	manager.client.Create(ctx, newPreCreatedSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		t.Error("Expected successful reconcile, got:", err)
 	}
 }
diff --git a/pkg/pki/pki_manager_test.go b/pkg/pki/pki_manager_test.go
index 62c30de8..af5da4f7 100644
--- a/pkg/pki/pki_manager_test.go
+++ b/pkg/pki/pki_manager_test.go
@@ -60,7 +60,7 @@ func TestGetPKIManager(t *testing.T) {
 
 	// Test mock functions
 	var err error
-	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		t.Error("Expected nil error got:", err)
 	}
 
diff --git a/pkg/pki/vaultpki/vault_test.go b/pkg/pki/vaultpki/vault_test.go
index 07609586..89b0d52d 100644
--- a/pkg/pki/vaultpki/vault_test.go
+++ b/pkg/pki/vaultpki/vault_test.go
@@ -164,7 +164,7 @@ func TestAll(t *testing.T) {
 		t.Fatal("Failed to convert test cert to JKS")
 	}
 
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err == nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err == nil {
 		t.Error("Expected resource not ready, got nil")
 	} else if reflect.TypeOf(err) != reflect.TypeOf(errorfactory.ResourceNotReady{}) {
 		t.Error("Expected resource not ready, got:", err)
@@ -188,7 +188,7 @@ func TestAll(t *testing.T) {
 	}))
 
 	// Should be safe to do multiple times
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		t.Error("Expected no error, got:", err)
 	}
 
@@ -197,7 +197,7 @@ func TestAll(t *testing.T) {
 		t.Error("Expected no error, got:", err)
 	}
 
-	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, make(map[string]v1beta1.ListenerStatusList)); err != nil {
 		t.Error("Expected no error, got:", err)
 	}
 
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index b8790a80..04bf7fab 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -27,26 +27,28 @@ import (
 
 func TestGenerateBrokerConfig(t *testing.T) {
 	tests := []struct {
-		testName                string
-		readOnlyConfig          string
-		zkAddresses             []string
-		zkPath                  string
-		kubernetesClusterDomain string
-		clusterWideConfig       string
-		perBrokerReadOnlyConfig string
-		perBrokerConfig         string
-		expectedConfig          string
-		perBrokerStorageConfig  []v1beta1.StorageConfig
+		testName                  string
+		readOnlyConfig            string
+		zkAddresses               []string
+		zkPath                    string
+		kubernetesClusterDomain   string
+		clusterWideConfig         string
+		perBrokerReadOnlyConfig   string
+		perBrokerConfig           string
+		advertisedListenerAddress string
+		expectedConfig            string
+		perBrokerStorageConfig    []v1beta1.StorageConfig
 	}{
 		{
-			testName:                "basicConfig",
-			readOnlyConfig:          ``,
-			zkAddresses:             []string{"example.zk:2181"},
-			zkPath:                  ``,
-			kubernetesClusterDomain: ``,
-			clusterWideConfig:       ``,
-			perBrokerConfig:         ``,
-			perBrokerReadOnlyConfig: ``,
+			testName:                  "basicConfig",
+			readOnlyConfig:            ``,
+			zkAddresses:               []string{"example.zk:2181"},
+			zkPath:                    ``,
+			kubernetesClusterDomain:   ``,
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			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
@@ -58,14 +60,15 @@ metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlM
 zookeeper.connect=example.zk:2181/`,
 		},
 		{
-			testName:                "basicConfigWithZKPath",
-			readOnlyConfig:          ``,
-			zkPath:                  `/kafka`,
-			kubernetesClusterDomain: ``,
-			zkAddresses:             []string{"example.zk:2181"},
-			clusterWideConfig:       ``,
-			perBrokerConfig:         ``,
-			perBrokerReadOnlyConfig: ``,
+			testName:                  "basicConfigWithZKPath",
+			readOnlyConfig:            ``,
+			zkPath:                    `/kafka`,
+			kubernetesClusterDomain:   ``,
+			zkAddresses:               []string{"example.zk:2181"},
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			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
@@ -77,14 +80,15 @@ metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlM
 zookeeper.connect=example.zk:2181/kafka`,
 		},
 		{
-			testName:                "basicConfigWithSimpleZkPath",
-			readOnlyConfig:          ``,
-			zkPath:                  `/`,
-			kubernetesClusterDomain: ``,
-			zkAddresses:             []string{"example.zk:2181"},
-			clusterWideConfig:       ``,
-			perBrokerConfig:         ``,
-			perBrokerReadOnlyConfig: ``,
+			testName:                  "basicConfigWithSimpleZkPath",
+			readOnlyConfig:            ``,
+			zkPath:                    `/`,
+			kubernetesClusterDomain:   ``,
+			zkAddresses:               []string{"example.zk:2181"},
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			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
@@ -96,14 +100,15 @@ metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlM
 zookeeper.connect=example.zk:2181/`,
 		},
 		{
-			testName:                "basicConfigWithMultipleZKAddressAndPath",
-			readOnlyConfig:          ``,
-			zkPath:                  `/kafka`,
-			zkAddresses:             []string{"example.zk:2181", "example.zk-1:2181"},
-			kubernetesClusterDomain: ``,
-			clusterWideConfig:       ``,
-			perBrokerConfig:         ``,
-			perBrokerReadOnlyConfig: ``,
+			testName:                  "basicConfigWithMultipleZKAddressAndPath",
+			readOnlyConfig:            ``,
+			zkPath:                    `/kafka`,
+			zkAddresses:               []string{"example.zk:2181", "example.zk-1:2181"},
+			kubernetesClusterDomain:   ``,
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			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
@@ -128,6 +133,7 @@ zookeeper.connect=example.zk:2181,example.zk-1:2181/kafka`,
 					MountPath: "/kafka-logs",
 				},
 			},
+			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
@@ -140,14 +146,15 @@ metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlM
 zookeeper.connect=example.zk:2181/`,
 		},
 		{
-			testName:                "basicConfigWithClusterDomain",
-			readOnlyConfig:          ``,
-			zkAddresses:             []string{"example.zk:2181"},
-			zkPath:                  ``,
-			kubernetesClusterDomain: `foo.bar`,
-			clusterWideConfig:       ``,
-			perBrokerConfig:         ``,
-			perBrokerReadOnlyConfig: ``,
+			testName:                  "basicConfigWithClusterDomain",
+			readOnlyConfig:            ``,
+			zkAddresses:               []string{"example.zk:2181"},
+			zkPath:                    ``,
+			kubernetesClusterDomain:   `foo.bar`,
+			clusterWideConfig:         ``,
+			perBrokerConfig:           ``,
+			perBrokerReadOnlyConfig:   ``,
+			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
@@ -175,6 +182,7 @@ compression.type=snappy
 			perBrokerReadOnlyConfig: `
 auto.create.topics.enable=true
 `,
+			advertisedListenerAddress: `kafka-0.kafka.svc.cluster.local:9092`,
 			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.cluster.local:9092
 auto.create.topics.enable=true
 broker.id=0
@@ -207,14 +215,14 @@ zookeeper.connect=example.zk:2181/`,
 							ZKAddresses: test.zkAddresses,
 							ZKPath:      test.zkPath,
 							ListenersConfig: v1beta1.ListenersConfig{
-								InternalListeners: []v1beta1.InternalListenerConfig{
-									{CommonListenerSpec: v1beta1.CommonListenerSpec{
+								InternalListeners: []v1beta1.InternalListenerConfig{{
+									CommonListenerSpec: v1beta1.CommonListenerSpec{
 										Type:          "plaintext",
 										Name:          "internal",
 										ContainerPort: 9092,
 									},
-										UsedForInnerBrokerCommunication: true,
-									},
+									UsedForInnerBrokerCommunication: true,
+								},
 								},
 							},
 							ReadOnlyConfig:          test.readOnlyConfig,
@@ -233,7 +241,17 @@ zookeeper.connect=example.zk:2181/`,
 					},
 				},
 			}
-			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, map[string]string{}, "", "", []string{}, logf.NullLogger{})
+
+			controllerListenerStatus := map[string]v1beta1.ListenerStatusList{
+				"internal": {
+					{
+						Name:    "broker-0",
+						Address: test.advertisedListenerAddress,
+					},
+				},
+			}
+
+			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, map[string]v1beta1.ListenerStatusList{}, map[string]v1beta1.ListenerStatusList{}, controllerListenerStatus, "", "", []string{}, logf.NullLogger{})
 
 			if generatedConfig != test.expectedConfig {
 				t.Errorf("the expected config is %s, received: %s", test.expectedConfig, generatedConfig)
diff --git a/pkg/util/client/common_test.go b/pkg/util/client/common_test.go
new file mode 100644
index 00000000..96abef6f
--- /dev/null
+++ b/pkg/util/client/common_test.go
@@ -0,0 +1,67 @@
+// Copyright © 2020 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 client
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+)
+
+func TestGenerateKafkaAddress(t *testing.T) {
+	cluster := &v1beta1.KafkaCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			Namespace: "test",
+		},
+		Spec: v1beta1.KafkaClusterSpec{
+			HeadlessServiceEnabled: true,
+			ListenersConfig: v1beta1.ListenersConfig{
+				InternalListeners: []v1beta1.InternalListenerConfig{
+					{
+						CommonListenerSpec: v1beta1.CommonListenerSpec{ContainerPort: 80},
+					},
+				},
+			},
+		},
+	}
+
+	generatedHeadlessWithoutPort := GenerateKafkaAddressWithoutPort(cluster)
+	expected := "test-headless.test.svc.cluster.local"
+	if generatedHeadlessWithoutPort != expected {
+		t.Error("Expected kafka address:", expected, "Got:", generatedHeadlessWithoutPort)
+	}
+
+	generatedHeadless := GenerateKafkaAddress(cluster)
+	expected = "test-headless.test.svc.cluster.local:80"
+	if generatedHeadless != expected {
+		t.Error("Expected kafka address:", expected, "Got:", generatedHeadless)
+	}
+
+	cluster.Spec.HeadlessServiceEnabled = false
+	generatedAllBrokerWithoutPort := GenerateKafkaAddressWithoutPort(cluster)
+	expected = "test-all-broker.test.svc.cluster.local"
+	if generatedAllBrokerWithoutPort != expected {
+		t.Error("Expected kafka address:", expected, "Got:", generatedAllBrokerWithoutPort)
+	}
+
+	generatedAllBroker := GenerateKafkaAddress(cluster)
+	expected = "test-all-broker.test.svc.cluster.local:80"
+	if generatedAllBroker != expected {
+		t.Error("Expected kafka address:", expected, "Got:", generatedAllBroker)
+	}
+}
diff --git a/pkg/util/pki/pki_common_test.go b/pkg/util/pki/pki_common_test.go
index 116f23ab..2e937521 100644
--- a/pkg/util/pki/pki_common_test.go
+++ b/pkg/util/pki/pki_common_test.go
@@ -130,7 +130,7 @@ func TestGetInternalDNSNames(t *testing.T) {
 
 func TestBrokerUserForCluster(t *testing.T) {
 	cluster := testCluster(t)
-	user := BrokerUserForCluster(cluster, map[string]string{})
+	user := BrokerUserForCluster(cluster, make(map[string]v1beta1.ListenerStatusList))
 
 	expected := &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster),