Repository: banzaicloud/kafka-operator
Pull request #505 (2020-11-23 15:46:00)

Multiple external listener
| Q               | A
| --------------- | ---
| Bug fix?        | no|
| New feature?    |yes|
| API breaks?     | no|
| Deprecations?   | no|
| License         | Apache 2.0


### What's in this PR?
<!-- Explain the contents of the PR. Give an overview about the implementation, which decisions were made and why. -->

This PR adds support for multiple external listener endpoints for Kafka.
This is a long limitation of the operator which allowed only one external listener per cluster.
Since Kafka is not handling advertised.listeners config as a read-only config any more this listener setting will only work when creating a new cluster with the operator. Adding a new listener to a running cluster will not work since it will trigger a rolling update which will fail. We are working on a different PR which treats these settings as a dynamic config to overcome this limitation. Also we know that we should allow to support multiple settings per listener, that will come soon as well.


### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [x] Implementation tested
- [x] Error handling code meets the [guideline](https://github.com/banzaicloud/developer-guide/blob/master/docs/coding-style/error-handling-guide.md)
- [x] Logging code meets the guideline
- [x] User guide and development docs updated (if needed)



Patch:
diff --git a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml b/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
index 2b2e2c42..f8fa9e8b 100644
--- a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
+++ b/config/samples/banzaicloud_v1beta1_kafkacluster.yaml
@@ -63,6 +63,9 @@ spec:
       # which has type read-only these config changes will trigger rolling upgrade
       readOnlyConfig: |
         auto.create.topics.enable=false
+        cruise.control.metrics.topic.auto.create=true
+        cruise.control.metrics.topic.num.partitions=1
+        cruise.control.metrics.topic.replication.factor=2
       brokerConfig:
         # Docker image used by the operator to create the Broker with id 0
         #image: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
diff --git a/config/samples/kafkacluster-with-istio.yaml b/config/samples/kafkacluster-with-istio.yaml
index 071b764a..490a9b28 100644
--- a/config/samples/kafkacluster-with-istio.yaml
+++ b/config/samples/kafkacluster-with-istio.yaml
@@ -16,6 +16,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokerConfigGroups:
     default:
       brokerAnnotations:
diff --git a/config/samples/kafkacluster_with_external_ssl.yaml b/config/samples/kafkacluster_with_external_ssl.yaml
index 257f70d4..65c79aa7 100644
--- a/config/samples/kafkacluster_with_external_ssl.yaml
+++ b/config/samples/kafkacluster_with_external_ssl.yaml
@@ -12,6 +12,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokers:
     - id: 0
       brokerConfig:
diff --git a/config/samples/kafkacluster_with_internal_ssl_vault.yaml b/config/samples/kafkacluster_with_internal_ssl_vault.yaml
index 2391d313..4bee3b0f 100644
--- a/config/samples/kafkacluster_with_internal_ssl_vault.yaml
+++ b/config/samples/kafkacluster_with_internal_ssl_vault.yaml
@@ -9,9 +9,12 @@ spec:
   zkAddresses:
     - "zookeeper-client.zookeeper:2181"
   oneBrokerPerNode: false
-  clusterImage: "wurstmeister/kafka:2.12-2.1.0"
+  clusterImage: "banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
     authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
     allow.everyone.if.no.acl.found=false
     ssl.secure.random.implementation=SHA1PRNG
diff --git a/config/samples/kafkacluster_with_ssl_groups.yaml b/config/samples/kafkacluster_with_ssl_groups.yaml
index f03ee767..1f7b8304 100644
--- a/config/samples/kafkacluster_with_ssl_groups.yaml
+++ b/config/samples/kafkacluster_with_ssl_groups.yaml
@@ -17,6 +17,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
     authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
     allow.everyone.if.no.acl.found=false
   brokerConfigGroups:
diff --git a/config/samples/kafkacluster_without_ssl.yaml b/config/samples/kafkacluster_without_ssl.yaml
index 44584714..67dfc05c 100644
--- a/config/samples/kafkacluster_without_ssl.yaml
+++ b/config/samples/kafkacluster_without_ssl.yaml
@@ -12,6 +12,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokers:
     - id: 0
       brokerConfig:
diff --git a/config/samples/kafkacluster_without_ssl_groups.yaml b/config/samples/kafkacluster_without_ssl_groups.yaml
index ce637c98..69944cbf 100644
--- a/config/samples/kafkacluster_without_ssl_groups.yaml
+++ b/config/samples/kafkacluster_without_ssl_groups.yaml
@@ -12,6 +12,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokerConfigGroups:
     default:
       storageConfigs:
diff --git a/config/samples/simplekafkacluster.yaml b/config/samples/simplekafkacluster.yaml
index 2c4f2523..938986ef 100644
--- a/config/samples/simplekafkacluster.yaml
+++ b/config/samples/simplekafkacluster.yaml
@@ -13,6 +13,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokerConfigGroups:
     default:
       storageConfigs:
diff --git a/config/samples/simplekafkacluster_ebs_csi.yaml b/config/samples/simplekafkacluster_ebs_csi.yaml
index bbdecd2a..3e8a9814 100644
--- a/config/samples/simplekafkacluster_ebs_csi.yaml
+++ b/config/samples/simplekafkacluster_ebs_csi.yaml
@@ -18,6 +18,9 @@ spec:
   clusterImage: "ghcr.io/banzaicloud/kafka:2.13-2.6.0-bzc.1"
   readOnlyConfig: |
     auto.create.topics.enable=false
+    cruise.control.metrics.topic.auto.create=true
+    cruise.control.metrics.topic.num.partitions=1
+    cruise.control.metrics.topic.replication.factor=2
   brokerConfigGroups:
     default:
       storageConfigs:
diff --git a/go.sum b/go.sum
index 1ed951ca..28cc75d6 100644
--- a/go.sum
+++ b/go.sum
@@ -668,7 +668,6 @@ github.com/jefferai/isbadcipher v0.0.0-20190226160619-51d2077c035f h1:E87tDTVS5W
 github.com/jefferai/isbadcipher v0.0.0-20190226160619-51d2077c035f/go.mod h1:3J2qVK16Lq8V+wfiL2lPeDZ7UWMxk5LemerHa1p6N00=
 github.com/jefferai/jsonx v1.0.0 h1:Xoz0ZbmkpBvED5W9W1B5B/zc3Oiq7oXqiW7iRV3B6EI=
 github.com/jefferai/jsonx v1.0.0/go.mod h1:OGmqmi2tTeI/PS+qQfBDToLHHJIy/RMp24fPo8vFvoQ=
-github.com/jessevdk/go-flags v1.4.0/go.mod h1:4FA24M0QyGHXBuZZK/XkWh8h0e1EYbRYJSGM75WSRxI=
 github.com/jetstack/cert-manager v0.15.2 h1:3P2d0aV0j7hOb5/QK2tSwWHQITb/QQEizGqqdoq+lD4=
 github.com/jetstack/cert-manager v0.15.2/go.mod h1:7V2UW1EzgIWVUWi4uVATMIWXqinFOEqpggdvFdNMhlk=
 github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af h1:pmfjZENx5imkbgOkpRUYLnmbU7UEFbjtDA2hxJ1ichM=
@@ -1436,8 +1435,6 @@ k8s.io/kube-openapi v0.0.0-20200121204235-bf4fb3bd569c h1:/KUFqjjqAcY4Us6luF5RDN
 k8s.io/kube-openapi v0.0.0-20200121204235-bf4fb3bd569c/go.mod h1:GRQhZsXIAJ1xR0C9bd8UpWHZ5plfAS9fzPjJuQ6JL3E=
 k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6 h1:Oh3Mzx5pJ+yIumsAD0MOECPVeXsVot0UkiaCGVyfGQY=
 k8s.io/kube-openapi v0.0.0-20200410145947-61e04a5be9a6/go.mod h1:GRQhZsXIAJ1xR0C9bd8UpWHZ5plfAS9fzPjJuQ6JL3E=
-k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29 h1:NeQXVJ2XFSkRoPzRo8AId01ZER+j8oV4SZADT4iBOXQ=
-k8s.io/kube-openapi v0.0.0-20200410145947-bcb3869e6f29/go.mod h1:F+5wygcW0wmRTnM3cOgIqGivxkwSWIWT5YdsDbeAOaU=
 k8s.io/kubectl v0.18.0/go.mod h1:LOkWx9Z5DXMEg5KtOjHhRiC1fqJPLyCr3KtQgEolCkU=
 k8s.io/metrics v0.18.0/go.mod h1:8aYTW18koXqjLVKL7Ds05RPMX9ipJZI3mywYvBOxXd4=
 k8s.io/utils v0.0.0-20191114184206-e782cd3c129f h1:GiPwtSzdP43eI1hpPCbROQCCIgCuiMMNF8YUVLF3vJo=
@@ -1446,8 +1443,6 @@ k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89 h1:d4vVOjXm687F1iLSP2q3lyPPuyvTU
 k8s.io/utils v0.0.0-20200324210504-a9aa75ae1b89/go.mod h1:sZAwmy6armz5eXlNoLmJcl4F1QuKu7sr+mFQ0byX7Ew=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7MpmSnvtrOieolJKoE=
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19 h1:7Nu2dTj82c6IaWvL7hImJzcXoTPz1MsSCH7r+0m6rfo=
-k8s.io/utils v0.0.0-20200619165400-6e3d28b6ed19/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 layeh.com/radius v0.0.0-20190322222518-890bc1058917 h1:BDXFaFzUt5EIqe/4wrTc4AcYZWP6iC6Ult+jQWLh5eU=
 layeh.com/radius v0.0.0-20190322222518-890bc1058917/go.mod h1:fywZKyu//X7iRzaxLgPWsvc0L26IUpVvE/aeIL2JtIQ=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
@@ -1467,7 +1462,6 @@ sigs.k8s.io/kustomize v2.0.3+incompatible/go.mod h1:MkjgH3RdOWrievjo6c9T245dYlB5
 sigs.k8s.io/structured-merge-diff v0.0.0-20190525122527-15d366b2352e/go.mod h1:wWxsB5ozmmv/SG7nM11ayaAW51xMvak/t1r0CSlcokI=
 sigs.k8s.io/structured-merge-diff v1.0.1-0.20191108220359-b1b620dd3f06 h1:zD2IemQ4LmOcAumeiyDWXKUI2SO0NYDe3H6QGvPOVgU=
 sigs.k8s.io/structured-merge-diff v1.0.1-0.20191108220359-b1b620dd3f06/go.mod h1:/ULNhyfzRopfcjskuui0cTITekDduZ7ycKN3oUT9R18=
-sigs.k8s.io/structured-merge-diff/v2 v2.0.1/go.mod h1:Wb7vfKAodbKgf6tn1Kl0VvGj7mRH6DGaRcixXEJXTsE=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0-20200116222232-67a7b8c61874/go.mod h1:PlARxl6Hbt/+BC80dRLi1qAmnMqwqDg62YvvVkZjemw=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0 h1:dOmIZBMfhcHS09XZkMyUgkq5trg3/jRyJYFZUiaOp8E=
 sigs.k8s.io/structured-merge-diff/v3 v3.0.0/go.mod h1:PlARxl6Hbt/+BC80dRLi1qAmnMqwqDg62YvvVkZjemw=
diff --git a/pkg/pki/certmanagerpki/certmanager_pki.go b/pkg/pki/certmanagerpki/certmanager_pki.go
index 86adfbca..8a606053 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki.go
@@ -18,11 +18,6 @@ import (
 	"context"
 	"fmt"
 
-	"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"
 	"github.com/go-logr/logr"
 	certv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1alpha2"
 	certmeta "github.com/jetstack/cert-manager/pkg/apis/meta/v1"
@@ -33,6 +28,12 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"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"
 )
 
 func (c *certManager) FinalizePKI(ctx context.Context, logger logr.Logger) error {
@@ -89,7 +90,7 @@ 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 []string) (err error) {
+func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) (err error) {
 	logger.Info("Reconciling cert-manager PKI")
 
 	resources, err := c.kafkapki(ctx, scheme, externalHostnames)
@@ -106,7 +107,7 @@ func (c *certManager) ReconcilePKI(ctx context.Context, logger logr.Logger, sche
 	return nil
 }
 
-func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, externalHostnames []string) ([]runtime.Object, error) {
+func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, externalHostnames map[string]string) ([]runtime.Object, error) {
 	sslConfig := c.cluster.Spec.ListenersConfig.SSLSecrets
 	if sslConfig.Create {
 		if sslConfig.IssuerRef == nil {
@@ -117,7 +118,7 @@ func (c *certManager) kafkapki(ctx context.Context, scheme *runtime.Scheme, exte
 	return userProvidedPKI(ctx, c.client, c.cluster, scheme, externalHostnames)
 }
 
-func userProvidedIssuerPKI(cluster *v1beta1.KafkaCluster, externalHostnames []string) []runtime.Object {
+func userProvidedIssuerPKI(cluster *v1beta1.KafkaCluster, externalHostnames map[string]string) []runtime.Object {
 	// No need to generate self-signed certs and issuers because the issuer is provided by user
 	return []runtime.Object{
 		// Broker "user"
@@ -127,7 +128,7 @@ func userProvidedIssuerPKI(cluster *v1beta1.KafkaCluster, externalHostnames []st
 	}
 }
 
-func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames []string) []runtime.Object {
+func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames map[string]string) []runtime.Object {
 	return []runtime.Object{
 		// A self-signer for the CA Certificate
 		selfSignerForCluster(cluster, scheme),
@@ -135,7 +136,7 @@ func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHost
 		caCertForCluster(cluster),
 		// A cluster issuer backed by the CA certificate - so it can provision secrets
 		// for producers/consumers in other namespaces
-		mainIssuerForCluster(cluster, scheme),
+		mainIssuerForCluster(cluster),
 		// Broker "user"
 		pkicommon.BrokerUserForCluster(cluster, externalHostnames),
 		// Operator user
@@ -143,7 +144,9 @@ func fullPKI(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHost
 	}
 }
 
-func userProvidedPKI(ctx context.Context, client client.Client, cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames []string) ([]runtime.Object, error) {
+func userProvidedPKI(
+	ctx context.Context, client client.Client,
+	cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme, externalHostnames map[string]string) ([]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 {
@@ -151,7 +154,7 @@ func userProvidedPKI(ctx context.Context, client client.Client, cluster *v1beta1
 	}
 	return []runtime.Object{
 		caSecret,
-		mainIssuerForCluster(cluster, scheme),
+		mainIssuerForCluster(cluster),
 		// The client/peer certificates in the secret will still work, however are not actually used.
 		// This will also make sure that if the peerCert/clientCert provided are invalid
 		// a valid one will still be used with the provided CA.
@@ -228,7 +231,7 @@ func caCertForCluster(cluster *v1beta1.KafkaCluster) *certv1.Certificate {
 	}
 }
 
-func mainIssuerForCluster(cluster *v1beta1.KafkaCluster, scheme *runtime.Scheme) *certv1.ClusterIssuer {
+func mainIssuerForCluster(cluster *v1beta1.KafkaCluster) *certv1.ClusterIssuer {
 	clusterIssuerMeta := templates.ObjectMeta(
 		fmt.Sprintf(pkicommon.BrokerClusterIssuerTemplate, cluster.Namespace, cluster.Name),
 		pkicommon.LabelsForKafkaPKI(cluster.Name, cluster.Namespace), cluster)
diff --git a/pkg/pki/pki_manager.go b/pkg/pki/pki_manager.go
index da77f626..b3c4e76c 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 []string) error {
+func (m *mockPKIManager) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) error {
 	return nil
 }
 
diff --git a/pkg/pki/vaultpki/vault_pki.go b/pkg/pki/vaultpki/vault_pki.go
index 5ac7954e..b49f3f81 100644
--- a/pkg/pki/vaultpki/vault_pki.go
+++ b/pkg/pki/vaultpki/vault_pki.go
@@ -19,10 +19,6 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
-	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
-	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
-	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
 	"github.com/go-logr/logr"
 	vaultapi "github.com/hashicorp/vault/api"
 	corev1 "k8s.io/api/core/v1"
@@ -30,9 +26,14 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+
+	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
+	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
+	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
+	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
 )
 
-func (v *vaultPKI) FinalizePKI(ctx context.Context, logger logr.Logger) error {
+func (v *vaultPKI) FinalizePKI(_ context.Context, _ logr.Logger) error {
 	vault, err := v.getClient()
 	if err != nil {
 		return err
@@ -46,7 +47,7 @@ func (v *vaultPKI) FinalizePKI(ctx context.Context, logger logr.Logger) error {
 	}
 	for _, user := range users {
 		if user != "broker" && user != "controller" {
-			err = fmt.Errorf("Non broker/controller user exists: %s", user)
+			err = fmt.Errorf("non broker/controller user exists: %s", user)
 			return errorfactory.New(errorfactory.ResourceNotReady{}, err, "pki still contains user certificates")
 		}
 	}
@@ -55,7 +56,7 @@ func (v *vaultPKI) FinalizePKI(ctx context.Context, logger 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 []string) (err error) {
+func (v *vaultPKI) ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) (err error) {
 	log := logger.WithName("vault_pki")
 
 	log.Info("Retrieving vault client")
@@ -151,7 +152,8 @@ func (v *vaultPKI) reconcileBootstrapSecrets(ctx context.Context, scheme *runtim
 	return
 }
 
-func (v *vaultPKI) reconcileBrokerCert(ctx context.Context, vault *vaultapi.Client, externalHostnames []string) (*pkicommon.UserCertificate, error) {
+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))
 }
 
diff --git a/pkg/resources/cruisecontrol/cruisecontrol.go b/pkg/resources/cruisecontrol/cruisecontrol.go
index 1a6d94a6..8149812a 100644
--- a/pkg/resources/cruisecontrol/cruisecontrol.go
+++ b/pkg/resources/cruisecontrol/cruisecontrol.go
@@ -84,10 +84,10 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	if r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint == "" {
 
-		err := generateCCTopic(r.KafkaCluster, r.Client, log)
-		if err != nil {
-			k8sutil.UpdateCRStatus(r.Client, r.KafkaCluster, v1beta1.CruiseControlTopicNotReady, log)
-			return err
+		genErr := generateCCTopic(r.KafkaCluster, r.Client, log.WithName("generateCCTopic"))
+		if genErr != nil {
+			updateErr := k8sutil.UpdateCRStatus(r.Client, r.KafkaCluster, v1beta1.CruiseControlTopicNotReady, log)
+			return errors.Combine(genErr, updateErr)
 		}
 		statusErr := k8sutil.UpdateCRStatus(r.Client, r.KafkaCluster, v1beta1.CruiseControlTopicReady, log)
 		if statusErr != nil {
diff --git a/pkg/resources/cruisecontrol/topicManager.go b/pkg/resources/cruisecontrol/topicManager.go
index 4e4acca6..ff590c8b 100644
--- a/pkg/resources/cruisecontrol/topicManager.go
+++ b/pkg/resources/cruisecontrol/topicManager.go
@@ -17,11 +17,13 @@ package cruisecontrol
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
+	"github.com/banzaicloud/kafka-operator/pkg/util"
 	"github.com/banzaicloud/kafka-operator/pkg/webhook"
 	"github.com/go-logr/logr"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -30,6 +32,7 @@ import (
 )
 
 const (
+	ccMetricTopicAutoCreate             = "cruise.control.metrics.topic.auto.create"
 	cruiseControlTopicFormat            = "%s-cruise-control-topic"
 	cruiseControlTopicName              = "__CruiseControlMetrics"
 	cruiseControlTopicPartitions        = 12
@@ -68,6 +71,19 @@ func newCruiseControlTopic(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaTopic {
 }
 
 func generateCCTopic(cluster *v1beta1.KafkaCluster, client client.Client, log logr.Logger) error {
+	readOnlyConfigProperties := util.ParsePropertiesFormat(cluster.Spec.ReadOnlyConfig)
+
+	// for compatibility reasons the only case when we let CC to create its own kafka topics is
+	// when we enable the creation explicitly
+	if propertyValue, present := readOnlyConfigProperties[ccMetricTopicAutoCreate]; present {
+		if autoCreate, err := strconv.ParseBool(propertyValue); err != nil {
+			return err
+		} else if autoCreate {
+			log.Info("CruiseControl topic has been created by CruiseControl")
+			return nil
+		}
+	}
+
 	existing := &v1alpha1.KafkaTopic{}
 	topic := newCruiseControlTopic(cluster)
 	if err := client.Get(context.TODO(), types.NamespacedName{Name: topic.Name, Namespace: topic.Namespace}, existing); err != nil {
@@ -90,5 +106,7 @@ func generateCCTopic(cluster *v1beta1.KafkaCluster, client client.Client, log lo
 			return errorfactory.New(errorfactory.APIFailure{}, err, "failed to lookup cruise control topic")
 		}
 	}
+
+	log.Info("CruiseControl topic has been created by Operator")
 	return nil
 }
diff --git a/pkg/resources/envoy/configmap.go b/pkg/resources/envoy/configmap.go
index afb78f0f..5d7c8041 100644
--- a/pkg/resources/envoy/configmap.go
+++ b/pkg/resources/envoy/configmap.go
@@ -36,10 +36,12 @@ import (
 	corev1 "k8s.io/api/core/v1"
 )
 
-func (r *Reconciler) configMap(log logr.Logger) runtime.Object {
+func (r *Reconciler) configMap(log logr.Logger, extListener v1beta1.ExternalListenerConfig) runtime.Object {
 	configMap := &corev1.ConfigMap{
-		ObjectMeta: templates.ObjectMeta(envoyVolumeAndConfigName, labelSelector, r.KafkaCluster),
-		Data:       map[string]string{"envoy.yaml": GenerateEnvoyConfig(r.KafkaCluster, log)},
+		ObjectMeta: templates.ObjectMeta(
+			fmt.Sprintf(envoyVolumeAndConfigName, extListener.Name, r.KafkaCluster.GetName()),
+			labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name), r.KafkaCluster),
+		Data: map[string]string{"envoy.yaml": GenerateEnvoyConfig(r.KafkaCluster, extListener, log)},
 	}
 	return configMap
 }
@@ -52,8 +54,7 @@ func generateAddressValue(kc *v1beta1.KafkaCluster, brokerId int) string {
 	return fmt.Sprintf("%s-%d.%s.svc.%s", kc.Name, brokerId, kc.Namespace, kc.Spec.GetKubernetesClusterDomain())
 }
 
-func GenerateEnvoyConfig(kc *v1beta1.KafkaCluster, log logr.Logger) string {
-	//TODO support multiple external listener by removing [0] (baluchicken)
+func GenerateEnvoyConfig(kc *v1beta1.KafkaCluster, elistener v1beta1.ExternalListenerConfig, log logr.Logger) string {
 	adminConfig := envoybootstrap.Admin{
 		AccessLogPath: "/tmp/admin_access.log",
 		Address: &envoycore.Address{
@@ -78,7 +79,7 @@ func GenerateEnvoyConfig(kc *v1beta1.KafkaCluster, log logr.Logger) string {
 					SocketAddress: &envoycore.SocketAddress{
 						Address: "0.0.0.0",
 						PortSpecifier: &envoycore.SocketAddress_PortValue{
-							PortValue: uint32(kc.Spec.ListenersConfig.ExternalListeners[0].ExternalStartingPort + int32(brokerId)),
+							PortValue: uint32(elistener.ExternalStartingPort + int32(brokerId)),
 						},
 					},
 				},
@@ -114,7 +115,7 @@ func GenerateEnvoyConfig(kc *v1beta1.KafkaCluster, log logr.Logger) string {
 						SocketAddress: &envoycore.SocketAddress{
 							Address: generateAddressValue(kc, brokerId),
 							PortSpecifier: &envoycore.SocketAddress_PortValue{
-								PortValue: uint32(kc.Spec.ListenersConfig.ExternalListeners[0].ContainerPort),
+								PortValue: uint32(elistener.ContainerPort),
 							},
 						},
 					},
diff --git a/pkg/resources/envoy/deployment.go b/pkg/resources/envoy/deployment.go
index b4b7d541..8a3a05a0 100644
--- a/pkg/resources/envoy/deployment.go
+++ b/pkg/resources/envoy/deployment.go
@@ -19,27 +19,29 @@ import (
 	"encoding/hex"
 	"fmt"
 
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/runtime"
+
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
-	"github.com/go-logr/logr"
-	"k8s.io/apimachinery/pkg/runtime"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func (r *Reconciler) deployment(log logr.Logger) runtime.Object {
+func (r *Reconciler) deployment(log logr.Logger, extListener v1beta1.ExternalListenerConfig) runtime.Object {
 
-	exposedPorts := getExposedContainerPorts(r.KafkaCluster.Spec.ListenersConfig.ExternalListeners,
+	configMapName := fmt.Sprintf(envoyVolumeAndConfigName, extListener.Name, r.KafkaCluster.GetName())
+	exposedPorts := getExposedContainerPorts(extListener,
 		util.GetBrokerIdsFromStatusAndSpec(r.KafkaCluster.Status.BrokersState, r.KafkaCluster.Spec.Brokers, log))
 	volumes := []corev1.Volume{
 		{
-			Name: envoyVolumeAndConfigName,
+			Name: configMapName,
 			VolumeSource: corev1.VolumeSource{
 				ConfigMap: &corev1.ConfigMapVolumeSource{
-					LocalObjectReference: corev1.LocalObjectReference{Name: envoyVolumeAndConfigName},
+					LocalObjectReference: corev1.LocalObjectReference{Name: configMapName},
 					DefaultMode:          util.Int32Pointer(0644),
 				},
 			},
@@ -48,23 +50,25 @@ func (r *Reconciler) deployment(log logr.Logger) runtime.Object {
 
 	volumeMounts := []corev1.VolumeMount{
 		{
-			Name:      envoyVolumeAndConfigName,
+			Name:      configMapName,
 			MountPath: "/etc/envoy",
 			ReadOnly:  true,
 		},
 	}
 
 	return &appsv1.Deployment{
-		ObjectMeta: templates.ObjectMeta(envoyDeploymentName, labelSelector, r.KafkaCluster),
+		ObjectMeta: templates.ObjectMeta(
+			fmt.Sprintf(envoyDeploymentName, extListener.Name, r.KafkaCluster.GetName()),
+			labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name), r.KafkaCluster),
 		Spec: appsv1.DeploymentSpec{
 			Selector: &metav1.LabelSelector{
-				MatchLabels: labelSelector,
+				MatchLabels: labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name),
 			},
 			Replicas: util.Int32Pointer(r.KafkaCluster.Spec.EnvoyConfig.GetReplicas()),
 			Template: corev1.PodTemplateSpec{
 				ObjectMeta: metav1.ObjectMeta{
-					Labels:      labelSelector,
-					Annotations: generatePodAnnotations(r.KafkaCluster, log),
+					Labels:      labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name),
+					Annotations: generatePodAnnotations(r.KafkaCluster, extListener, log),
 				},
 				Spec: corev1.PodSpec{
 					ServiceAccountName: r.KafkaCluster.Spec.EnvoyConfig.GetServiceAccount(),
@@ -88,23 +92,22 @@ func (r *Reconciler) deployment(log logr.Logger) runtime.Object {
 	}
 }
 
-func getExposedContainerPorts(extListeners []v1beta1.ExternalListenerConfig, brokerIds []int) []corev1.ContainerPort {
+func getExposedContainerPorts(extListener v1beta1.ExternalListenerConfig, brokerIds []int) []corev1.ContainerPort {
 	var exposedPorts []corev1.ContainerPort
 
-	for _, eListener := range extListeners {
-		for _, id := range brokerIds {
-			exposedPorts = append(exposedPorts, corev1.ContainerPort{
-				Name:          fmt.Sprintf("broker-%d", id),
-				ContainerPort: eListener.ExternalStartingPort + int32(id),
-				Protocol:      corev1.ProtocolTCP,
-			})
-		}
+	for _, id := range brokerIds {
+		exposedPorts = append(exposedPorts, corev1.ContainerPort{
+			Name:          fmt.Sprintf("broker-%d", id),
+			ContainerPort: extListener.ExternalStartingPort + int32(id),
+			Protocol:      corev1.ProtocolTCP,
+		})
 	}
 	return exposedPorts
 }
 
-func generatePodAnnotations(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger) map[string]string {
-	hashedEnvoyConfig := sha256.Sum256([]byte(GenerateEnvoyConfig(kafkaCluster, log)))
+func generatePodAnnotations(kafkaCluster *v1beta1.KafkaCluster,
+	extListener v1beta1.ExternalListenerConfig, log logr.Logger) map[string]string {
+	hashedEnvoyConfig := sha256.Sum256([]byte(GenerateEnvoyConfig(kafkaCluster, extListener, log)))
 	annotations := map[string]string{
 		"envoy.yaml.hash": hex.EncodeToString(hashedEnvoyConfig[:]),
 	}
diff --git a/pkg/resources/envoy/envoy.go b/pkg/resources/envoy/envoy.go
index fb24b4f9..fc649dad 100644
--- a/pkg/resources/envoy/envoy.go
+++ b/pkg/resources/envoy/envoy.go
@@ -25,13 +25,16 @@ import (
 )
 
 const (
-	componentName            = "envoy"
-	envoyVolumeAndConfigName = "envoy-config"
-	envoyDeploymentName      = "envoy"
+	componentName = "envoy"
+	// The deployment and configmap name should made from the external listener name the cluster name to avoid all naming collision
+	envoyVolumeAndConfigName = "envoy-config-%s-%s"
+	envoyDeploymentName      = "envoy-%s-%s"
 )
 
-var labelSelector = map[string]string{
-	"app": "envoy",
+// labelsForEnvoyIngress returns the labels for selecting the resources
+// belonging to the given kafka CR name.
+func labelsForEnvoyIngress(crName, eLName string) map[string]string {
+	return map[string]string{"app": "envoyingress", "eListenerName": eLName, "kafka_cr": crName}
 }
 
 // Reconciler implements the Component Reconciler
@@ -56,15 +59,17 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 	log.V(1).Info("Reconciling")
 	if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners != nil && r.KafkaCluster.Spec.GetIngressController() == envoyutils.IngressControllerName {
 
-		for _, res := range []resources.ResourceWithLogs{
-			r.loadBalancer,
-			r.configMap,
-			r.deployment,
-		} {
-			o := res(log)
-			err := k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
-			if err != nil {
-				return err
+		for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
+			for _, res := range []resources.ResourceWithLogAndExternalListenerConfig{
+				r.loadBalancer,
+				r.configMap,
+				r.deployment,
+			} {
+				o := res(log, eListener)
+				err := k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
+				if err != nil {
+					return err
+				}
 			}
 		}
 	}
diff --git a/pkg/resources/envoy/loadbalancer.go b/pkg/resources/envoy/loadbalancer.go
index f12bfeb1..8c64d4a3 100644
--- a/pkg/resources/envoy/loadbalancer.go
+++ b/pkg/resources/envoy/loadbalancer.go
@@ -30,16 +30,18 @@ import (
 )
 
 // loadBalancer return a Loadbalancer service for Envoy
-func (r *Reconciler) loadBalancer(log logr.Logger) runtime.Object {
+func (r *Reconciler) loadBalancer(log logr.Logger, extListener v1beta1.ExternalListenerConfig) runtime.Object {
 
-	exposedPorts := getExposedServicePorts(r.KafkaCluster.Spec.ListenersConfig.ExternalListeners,
+	exposedPorts := getExposedServicePorts(extListener,
 		util.GetBrokerIdsFromStatusAndSpec(r.KafkaCluster.Status.BrokersState, r.KafkaCluster.Spec.Brokers, log))
 
 	service := &corev1.Service{
-		ObjectMeta: templates.ObjectMetaWithAnnotations(envoyutils.EnvoyServiceName, map[string]string{},
+		ObjectMeta: templates.ObjectMetaWithAnnotations(
+			fmt.Sprintf(envoyutils.EnvoyServiceName, extListener.Name, r.KafkaCluster.GetName()),
+			labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name),
 			r.KafkaCluster.Spec.EnvoyConfig.GetAnnotations(), r.KafkaCluster),
 		Spec: corev1.ServiceSpec{
-			Selector:                 map[string]string{"app": "envoy"},
+			Selector:                 labelsForEnvoyIngress(r.KafkaCluster.GetName(), extListener.Name),
 			Type:                     corev1.ServiceTypeLoadBalancer,
 			Ports:                    exposedPorts,
 			LoadBalancerSourceRanges: r.KafkaCluster.Spec.EnvoyConfig.GetLoadBalancerSourceRanges(),
@@ -49,18 +51,15 @@ func (r *Reconciler) loadBalancer(log logr.Logger) runtime.Object {
 	return service
 }
 
-func getExposedServicePorts(extListeners []v1beta1.ExternalListenerConfig, brokersIds []int) []corev1.ServicePort {
+func getExposedServicePorts(extListener v1beta1.ExternalListenerConfig, brokersIds []int) []corev1.ServicePort {
 	var exposedPorts []corev1.ServicePort
-
-	for _, eListener := range extListeners {
-		for _, brokerId := range brokersIds {
-			exposedPorts = append(exposedPorts, corev1.ServicePort{
-				Name:       fmt.Sprintf("broker-%d", brokerId),
-				Port:       eListener.ExternalStartingPort + int32(brokerId),
-				TargetPort: intstr.FromInt(int(eListener.ExternalStartingPort) + brokerId),
-				Protocol:   corev1.ProtocolTCP,
-			})
-		}
+	for _, brokerId := range brokersIds {
+		exposedPorts = append(exposedPorts, corev1.ServicePort{
+			Name:       fmt.Sprintf("broker-%d", brokerId),
+			Port:       extListener.ExternalStartingPort + int32(brokerId),
+			TargetPort: intstr.FromInt(int(extListener.ExternalStartingPort) + brokerId),
+			Protocol:   corev1.ProtocolTCP,
+		})
 	}
 	return exposedPorts
 }
diff --git a/pkg/resources/istioingress/istioingress.go b/pkg/resources/istioingress/istioingress.go
index f77a0275..a3154baf 100644
--- a/pkg/resources/istioingress/istioingress.go
+++ b/pkg/resources/istioingress/istioingress.go
@@ -34,7 +34,7 @@ const (
 // labelsForIstioIngress returns the labels for selecting the resources
 // belonging to the given kafka CR name.
 func labelsForIstioIngress(crName, eLName string) map[string]string {
-	return map[string]string{"app": "istioingress", "eListenerName": eLName}
+	return map[string]string{"app": "istioingress", "eListenerName": eLName, "kafka_cr": crName}
 }
 
 // Reconciler implements the Component Reconciler
diff --git a/pkg/resources/istioingress/meshgateway.go b/pkg/resources/istioingress/meshgateway.go
index 4fb01f79..8b78d4c3 100644
--- a/pkg/resources/istioingress/meshgateway.go
+++ b/pkg/resources/istioingress/meshgateway.go
@@ -31,7 +31,9 @@ import (
 
 func (r *Reconciler) meshgateway(log logr.Logger, externalListenerConfig v1beta1.ExternalListenerConfig) runtime.Object {
 	mgateway := &istioOperatorApi.MeshGateway{
-		ObjectMeta: templates.ObjectMeta(fmt.Sprintf(istioingressutils.MeshGatewayNameTemplate, r.KafkaCluster.Name), labelsForIstioIngress(r.KafkaCluster.Name, externalListenerConfig.Name), r.KafkaCluster),
+		ObjectMeta: templates.ObjectMeta(
+			fmt.Sprintf(istioingressutils.MeshGatewayNameTemplate, externalListenerConfig.Name, r.KafkaCluster.Name),
+			labelsForIstioIngress(r.KafkaCluster.Name, externalListenerConfig.Name), r.KafkaCluster),
 		Spec: istioOperatorApi.MeshGatewaySpec{
 			MeshGatewayConfiguration: istioOperatorApi.MeshGatewayConfiguration{
 				Labels:             labelsForIstioIngress(r.KafkaCluster.Name, externalListenerConfig.Name),
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index 9fbf2580..4c115b81 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -22,15 +22,16 @@ import (
 	"strings"
 	"text/template"
 
+	"github.com/go-logr/logr"
+	"github.com/imdario/mergo"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	zookeeperutils "github.com/banzaicloud/kafka-operator/pkg/util/zookeeper"
-	"github.com/go-logr/logr"
-	"github.com/imdario/mergo"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/runtime"
 )
 
 var kafkaConfigTemplate = `
@@ -75,7 +76,7 @@ super.users={{ .SuperUsers }}
 {{ end }}
 `
 
-func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, loadBalancerIPs map[string]string, 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{}{
@@ -109,7 +110,7 @@ func generateSuperUsers(users []string) (suStrings []string) {
 	return
 }
 
-func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) runtime.Object {
+func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs map[string]string, serverPass, clientPass string, superUsers []string, log logr.Logger) runtime.Object {
 	brokerConf := &corev1.ConfigMap{
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(brokerConfigTemplate+"-%d", r.KafkaCluster.Name, id),
@@ -127,12 +128,11 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loa
 	return brokerConf
 }
 
-func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIPs []string, domain, namespace, crName string, headlessServiceEnabled bool) string {
-	advertisedListenerConfig := []string{}
+func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIPs map[string]string, domain, namespace, crName string, headlessServiceEnabled bool) string {
+	advertisedListenerConfig := make([]string, 0, len(l.ExternalListeners)+len(l.InternalListeners))
 	for _, eListener := range l.ExternalListeners {
-		// use first element of loadBalancerIPs slice for external listener name
 		advertisedListenerConfig = append(advertisedListenerConfig,
-			fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), loadBalancerIPs[0], eListener.ExternalStartingPort+id))
+			fmt.Sprintf("%s://%s:%d", strings.ToUpper(eListener.Name), loadBalancerIPs[eListener.Name], eListener.ExternalStartingPort+id))
 	}
 	for _, iListener := range l.InternalListeners {
 		if headlessServiceEnabled {
@@ -147,7 +147,7 @@ func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadB
 }
 
 func generateStorageConfig(sConfig []v1beta1.StorageConfig) string {
-	mountPaths := []string{}
+	mountPaths := make([]string, 0, len(sConfig))
 	for _, storage := range sConfig {
 		mountPaths = append(mountPaths, storage.MountPath+`/kafka`)
 	}
@@ -213,7 +213,7 @@ func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32,
 	return internalListener
 }
 
-func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs []string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
+func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerConfig, loadBalancerIPs map[string]string, serverPass, clientPass string, superUsers []string, log logr.Logger) string {
 	parsedReadOnlyClusterConfig := util.ParsePropertiesFormat(r.KafkaCluster.Spec.ReadOnlyConfig)
 	var parsedReadOnlyBrokerConfig = map[string]string{}
 
@@ -239,7 +239,7 @@ func (r Reconciler) generateBrokerConfig(id int32, brokerConfig *v1beta1.BrokerC
 		log.Error(err, "error occurred during merging readOnly config to complete configs")
 	}
 
-	completeConfig := []string{}
+	completeConfig := make([]string, 0, len(completeConfigMap))
 
 	for key, value := range completeConfigMap {
 		completeConfig = append(completeConfig, fmt.Sprintf("%s=%s", key, value))
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index 6195c29a..8315cfd0 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -111,13 +111,13 @@ func getCreatedPvcForBroker(c client.Client, brokerID int32, namespace, crName s
 	return foundPvcList.Items, nil
 }
 
-func getLoadBalancerIP(client client.Client, namespace, ingressController, crName string, log logr.Logger) (string, error) {
+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, crName)
+		iControllerServiceName = fmt.Sprintf(istioingressutils.MeshGatewayNameTemplate, extListenerName, crName)
 	} else if ingressController == envoyutils.IngressControllerName {
-		iControllerServiceName = envoyutils.EnvoyServiceName
+		iControllerServiceName = fmt.Sprintf(envoyutils.EnvoyServiceName, extListenerName, crName)
 	}
 
 	err := client.Get(context.TODO(), types.NamespacedName{Name: iControllerServiceName, Namespace: namespace}, foundLBService)
@@ -195,21 +195,18 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		return errors.WrapIf(err, "failed to reconcile resource")
 	}
 
-	lbIPs := make([]string, 0)
+	lbIPs := make(map[string]string, len(r.KafkaCluster.Spec.ListenersConfig.ExternalListeners))
 
-	if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners != nil {
-		// TODO: This is a hack that needs to be banished when the time is right.
-		// Currently we only support one external listener but this will be fixed
-		// sometime in the future
-		if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners[0].HostnameOverride != "" {
-			// first element of slice will be used for external advertised listener
-			lbIPs = append(lbIPs, r.KafkaCluster.Spec.ListenersConfig.ExternalListeners[0].HostnameOverride)
+	for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
+		if eListener.HostnameOverride != "" {
+			lbIPs[eListener.Name] = eListener.HostnameOverride
 		}
-		lbIP, err := getLoadBalancerIP(r.Client, r.KafkaCluster.Namespace, r.KafkaCluster.Spec.GetIngressController(), r.KafkaCluster.Name, log)
+		lbIP, err := getLoadBalancerIP(r.Client, r.KafkaCluster.GetNamespace(),
+			r.KafkaCluster.Spec.GetIngressController(), r.KafkaCluster.GetName(), eListener.Name)
 		if err != nil {
 			return err
 		}
-		lbIPs = append(lbIPs, lbIP)
+		lbIPs[eListener.Name] = lbIP
 	}
 
 	// Setup the PKI if using SSL
@@ -508,7 +505,7 @@ func (r *Reconciler) reconcilePerBrokerDynamicConfig(brokerId int32, brokerConfi
 	// Calling DescribePerBrokerConfig with empty slice will return all config for that broker including the default ones
 	if len(parsedBrokerConfig) > 0 {
 
-		brokerConfigKeys := []string{}
+		brokerConfigKeys := make([]string, 0, len(parsedBrokerConfig))
 		for key := range parsedBrokerConfig {
 			brokerConfigKeys = append(brokerConfigKeys, key)
 		}
@@ -655,7 +652,7 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod)
 		//Since toleration does not support patchStrategy:"merge,retainKeys", we need to add all toleration from the current pod if the toleration is set in the CR
 		if len(desiredPod.Spec.Tolerations) > 0 {
 			desiredPod.Spec.Tolerations = append(desiredPod.Spec.Tolerations, currentPod.Spec.Tolerations...)
-			uniqueTolerations := []corev1.Toleration{}
+			uniqueTolerations := make([]corev1.Toleration, 0, len(desiredPod.Spec.Tolerations))
 			keys := make(map[corev1.Toleration]bool)
 			for _, t := range desiredPod.Spec.Tolerations {
 				if _, value := keys[t]; !value {
diff --git a/pkg/util/envoy/common.go b/pkg/util/envoy/common.go
index 95362b38..a6192f01 100644
--- a/pkg/util/envoy/common.go
+++ b/pkg/util/envoy/common.go
@@ -16,7 +16,7 @@ package envoy
 
 const (
 	// EnvoyServiceName name for loadbalancer service
-	EnvoyServiceName = "envoy-loadbalancer"
+	EnvoyServiceName = "envoy-loadbalancer-%s-%s"
 	// IngressControllerName name for envoy ingress service
 	IngressControllerName = "envoy"
 )
diff --git a/pkg/util/istioingress/common.go b/pkg/util/istioingress/common.go
index f915e609..8ee84b4a 100644
--- a/pkg/util/istioingress/common.go
+++ b/pkg/util/istioingress/common.go
@@ -18,5 +18,5 @@ const (
 	// IngressControllerName name for istioingress ingress service
 	IngressControllerName = "istioingress"
 	// IngressControllerName name for istioingress gateway service
-	MeshGatewayNameTemplate = "%s-meshgateway"
+	MeshGatewayNameTemplate = "meshgateway-%s-%s"
 )
diff --git a/pkg/util/pki/common.go b/pkg/util/pki/common.go
index 45f8dcb2..2471e657 100644
--- a/pkg/util/pki/common.go
+++ b/pkg/util/pki/common.go
@@ -18,14 +18,16 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"sort"
+
+	"github.com/go-logr/logr"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/resources/templates"
 	certutil "github.com/banzaicloud/kafka-operator/pkg/util/cert"
 	"github.com/banzaicloud/kafka-operator/pkg/util/kafka"
-	"github.com/go-logr/logr"
-	"k8s.io/apimachinery/pkg/runtime"
 )
 
 const (
@@ -44,8 +46,6 @@ const (
 	// BrokerControllerFQDNTemplate is combined with the above and cluster namespace
 	// to create a 'fake' full-name for the controller user
 	BrokerControllerFQDNTemplate = "%s.%s.mgt.%s"
-	// CAIntermediateTemplate is the template used for intermediate CA resources
-	CAIntermediateTemplate = "%s-intermediate.%s.%s"
 	// CAFQDNTemplate is the template used for the FQDN of a CA
 	CAFQDNTemplate = "%s-ca.%s.cluster.local"
 )
@@ -55,7 +55,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 []string) error
+	ReconcilePKI(ctx context.Context, logger logr.Logger, scheme *runtime.Scheme, externalHostnames map[string]string) error
 
 	// FinalizePKI performs any cleanup steps necessary for a PKI backend
 	FinalizePKI(ctx context.Context, logger logr.Logger) error
@@ -161,12 +161,17 @@ 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 []string) *v1alpha1.KafkaUser {
+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)
+	}
+	sort.Strings(additionalHosts)
 	return &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster), LabelsForKafkaPKI(cluster.Name, cluster.Namespace), cluster),
 		Spec: v1alpha1.KafkaUserSpec{
 			SecretName: fmt.Sprintf(BrokerServerCertTemplate, cluster.Name),
-			DNSNames:   append(GetInternalDNSNames(cluster), additionalHostnames...),
+			DNSNames:   append(GetInternalDNSNames(cluster), additionalHosts...),
 			IncludeJKS: true,
 			ClusterRef: v1alpha1.ClusterReference{
 				Name:      cluster.Name,


Test patch:
diff --git a/pkg/pki/certmanagerpki/certmanager_pki_test.go b/pkg/pki/certmanagerpki/certmanager_pki_test.go
index 828c70d6..bc277543 100644
--- a/pkg/pki/certmanagerpki/certmanager_pki_test.go
+++ b/pkg/pki/certmanagerpki/certmanager_pki_test.go
@@ -20,13 +20,14 @@ import (
 	"reflect"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"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"
-	corev1 "k8s.io/api/core/v1"
-	"k8s.io/client-go/kubernetes/scheme"
-	ctrl "sigs.k8s.io/controller-runtime"
 )
 
 var log = ctrl.Log.WithName("testing")
@@ -96,33 +97,33 @@ func TestReconcilePKI(t *testing.T) {
 	ctx := context.Background()
 
 	manager.client.Create(ctx, newServerSecret())
-	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err == nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err != nil {
+	if err := manager.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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 c48f86ef..62c30de8 100644
--- a/pkg/pki/pki_manager_test.go
+++ b/pkg/pki/pki_manager_test.go
@@ -19,11 +19,12 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
-	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/go-logr/logr"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
 
 var log logr.Logger
@@ -59,7 +60,7 @@ func TestGetPKIManager(t *testing.T) {
 
 	// Test mock functions
 	var err error
-	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme, []string{}); err != nil {
+	if err = mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); err != nil {
 		t.Error("Expected nil error got:", err)
 	}
 
@@ -89,7 +90,7 @@ func TestGetPKIManager(t *testing.T) {
 	}
 
 	// Default should be cert-manager also
-	cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend = v1beta1.PKIBackend("")
+	cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend = ""
 	certmanager = GetPKIManager(&mockClient{}, cluster, v1beta1.PKIBackendProvided)
 	pkiType = reflect.TypeOf(certmanager).String()
 	expected = "*certmanagerpki.certManager"
diff --git a/pkg/pki/vaultpki/vault_test.go b/pkg/pki/vaultpki/vault_test.go
index 06302af2..07609586 100644
--- a/pkg/pki/vaultpki/vault_test.go
+++ b/pkg/pki/vaultpki/vault_test.go
@@ -21,20 +21,20 @@ import (
 	"reflect"
 	"testing"
 
-	"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"
 	"github.com/hashicorp/vault/api"
 	"github.com/hashicorp/vault/builtin/logical/pki"
 	"github.com/hashicorp/vault/http"
 	"github.com/hashicorp/vault/sdk/logical"
 	"github.com/hashicorp/vault/vault"
-	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
+
+	"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"
 )
 
 var log = logf.Log.WithName("testing")
@@ -43,7 +43,7 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	cluster := &v1beta1.KafkaCluster{}
 	cluster.Name = "test"
 	cluster.Namespace = "test-namespace"
-	cluster.UID = types.UID("test-uid")
+	cluster.UID = "test-uid"
 	cluster.Spec = v1beta1.KafkaClusterSpec{}
 	cluster.Spec.ListenersConfig = v1beta1.ListenersConfig{}
 	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{
@@ -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, []string{}); err == nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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, []string{}); err != nil {
+	if err := mock.ReconcilePKI(ctx, log, scheme.Scheme, map[string]string{}); 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 6bc794cb..b8790a80 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -17,11 +17,12 @@ package kafka
 import (
 	"testing"
 
-	"github.com/banzaicloud/kafka-operator/api/v1beta1"
-	"github.com/banzaicloud/kafka-operator/pkg/resources"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/scheme"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
+	"github.com/banzaicloud/kafka-operator/pkg/resources"
 )
 
 func TestGenerateBrokerConfig(t *testing.T) {
@@ -232,7 +233,7 @@ zookeeper.connect=example.zk:2181/`,
 					},
 				},
 			}
-			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, []string{}, "", "", []string{}, logf.NullLogger{})
+			generatedConfig := r.generateBrokerConfig(0, r.KafkaCluster.Spec.Brokers[0].BrokerConfig, map[string]string{}, "", "", []string{}, logf.NullLogger{})
 
 			if generatedConfig != test.expectedConfig {
 				t.Errorf("the expected config is %s, received: %s", test.expectedConfig, generatedConfig)
diff --git a/pkg/util/pki/pki_common_test.go b/pkg/util/pki/pki_common_test.go
index 76296be3..0dec2ec6 100644
--- a/pkg/util/pki/pki_common_test.go
+++ b/pkg/util/pki/pki_common_test.go
@@ -122,7 +122,7 @@ func TestGetInternalDNSNames(t *testing.T) {
 
 func TestBrokerUserForCluster(t *testing.T) {
 	cluster := testCluster(t)
-	user := BrokerUserForCluster(cluster, []string{})
+	user := BrokerUserForCluster(cluster, map[string]string{})
 
 	expected := &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(GetCommonName(cluster),