Repository: banzaicloud/kafka-operator
Pull request #337 (2020-04-14 09:37:51)

Add pki manager / user support to operator
| 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 the ability to use different PKI/User by moving the PKI type and issuerRef to the User CRD.
Also it makes Kubernetes service names Istio friendly.


### 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/pipeline/blob/master/docs/error-handling-guide.md)
- [x] Logging code meets the guideline
- [x] User guide and development docs updated (if needed)



Patch:
diff --git a/api/v1alpha1/kafkauser_types.go b/api/v1alpha1/kafkauser_types.go
index 77798d5d..cba4494c 100644
--- a/api/v1alpha1/kafkauser_types.go
+++ b/api/v1alpha1/kafkauser_types.go
@@ -15,17 +15,28 @@
 package v1alpha1
 
 import (
+	cmmeta "github.com/jetstack/cert-manager/pkg/apis/meta/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 )
 
 // KafkaUserSpec defines the desired state of KafkaUser
 // +k8s:openapi-gen=true
 type KafkaUserSpec struct {
-	SecretName  string           `json:"secretName"`
-	ClusterRef  ClusterReference `json:"clusterRef"`
-	DNSNames    []string         `json:"dnsNames,omitempty"`
-	TopicGrants []UserTopicGrant `json:"topicGrants,omitempty"`
-	IncludeJKS  bool             `json:"includeJKS,omitempty"`
+	SecretName     string           `json:"secretName"`
+	ClusterRef     ClusterReference `json:"clusterRef"`
+	DNSNames       []string         `json:"dnsNames,omitempty"`
+	TopicGrants    []UserTopicGrant `json:"topicGrants,omitempty"`
+	IncludeJKS     bool             `json:"includeJKS,omitempty"`
+	CreateCert     *bool            `json:"createCert,omitempty"`
+	PKIBackendSpec PKIBackendSpec   `json:"pkiBackendSpec,omitempty"`
+}
+
+type PKIBackendSpec struct {
+	IssuerRef *cmmeta.ObjectReference `json:"issuerRef"`
+	// +kubebuilder:validation:Enum={"cert-manager","vault"}
+	PKIBackend v1beta1.PKIBackend `json:"pkiBackend"`
 }
 
 // UserTopicGrant is the desired permissions for the KafkaUser
@@ -69,3 +80,10 @@ type KafkaUserList struct {
 func init() {
 	SchemeBuilder.Register(&KafkaUser{}, &KafkaUserList{})
 }
+
+func (spec *KafkaUserSpec) GetIfCertShouldBeCreated() bool {
+	if spec.CreateCert != nil {
+		return *spec.CreateCert
+	}
+	return true
+}
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
index 78d31c52..a6f60390 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -19,6 +19,7 @@
 package v1alpha1
 
 import (
+	"github.com/jetstack/cert-manager/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -207,6 +208,12 @@ func (in *KafkaUserSpec) DeepCopyInto(out *KafkaUserSpec) {
 		*out = make([]UserTopicGrant, len(*in))
 		copy(*out, *in)
 	}
+	if in.CreateCert != nil {
+		in, out := &in.CreateCert, &out.CreateCert
+		*out = new(bool)
+		**out = **in
+	}
+	in.PKIBackendSpec.DeepCopyInto(&out.PKIBackendSpec)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new KafkaUserSpec.
@@ -239,6 +246,26 @@ func (in *KafkaUserStatus) DeepCopy() *KafkaUserStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *PKIBackendSpec) DeepCopyInto(out *PKIBackendSpec) {
+	*out = *in
+	if in.IssuerRef != nil {
+		in, out := &in.IssuerRef, &out.IssuerRef
+		*out = new(v1.ObjectReference)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PKIBackendSpec.
+func (in *PKIBackendSpec) DeepCopy() *PKIBackendSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(PKIBackendSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *UserTopicGrant) DeepCopyInto(out *UserTopicGrant) {
 	*out = *in
diff --git a/api/v1beta1/common_types.go b/api/v1beta1/common_types.go
index 361b2612..c00f4868 100644
--- a/api/v1beta1/common_types.go
+++ b/api/v1beta1/common_types.go
@@ -70,6 +70,9 @@ const (
 	PKIBackendCertManager PKIBackend = "cert-manager"
 	// PKIBackendVault invokes vault PKI for user certificate management
 	PKIBackendVault PKIBackend = "vault"
+	// PKIBackendProvided used to point the operator to use the PKI set in the cluster CR
+	// for admin and users required for the cluster to run
+	PKIBackendProvided PKIBackend = "pki-backend-provided"
 )
 
 // GracefulActionState holds information about GracefulAction State
diff --git a/api/v1beta1/kafkacluster_types.go b/api/v1beta1/kafkacluster_types.go
index c241dd72..d7dba504 100644
--- a/api/v1beta1/kafkacluster_types.go
+++ b/api/v1beta1/kafkacluster_types.go
@@ -225,20 +225,23 @@ type AlertManagerConfig struct {
 
 // ExternalListenerConfig defines the external listener config for Kafka
 type ExternalListenerConfig struct {
-	Type                 string `json:"type"`
-	Name                 string `json:"name"`
+	CommonListenerSpec   `json:",inline"`
 	ExternalStartingPort int32  `json:"externalStartingPort"`
-	ContainerPort        int32  `json:"containerPort"`
 	HostnameOverride     string `json:"hostnameOverride,omitempty"`
 }
 
 // InternalListenerConfig defines the internal listener config for Kafka
 type InternalListenerConfig struct {
-	Type                            string `json:"type"`
-	Name                            string `json:"name"`
-	UsedForInnerBrokerCommunication bool   `json:"usedForInnerBrokerCommunication"`
-	ContainerPort                   int32  `json:"containerPort"`
-	UsedForControllerCommunication  bool   `json:"usedForControllerCommunication,omitempty"`
+	CommonListenerSpec              `json:",inline"`
+	UsedForInnerBrokerCommunication bool `json:"usedForInnerBrokerCommunication"`
+	UsedForControllerCommunication  bool `json:"usedForControllerCommunication,omitempty"`
+}
+
+// CommonListenerSpec defines the common building block for Listener type
+type CommonListenerSpec struct {
+	Type          string `json:"type"`
+	Name          string `json:"name"`
+	ContainerPort int32  `json:"containerPort"`
 }
 
 // +kubebuilder:object:root=true
@@ -283,6 +286,14 @@ func (iIConfig *IstioIngressConfig) GetResources() *corev1.ResourceRequirements
 	}
 }
 
+// GetListenerName returns the prepared listener name
+func (lP *CommonListenerSpec) GetListenerServiceName() string {
+	if !strings.HasPrefix(lP.Name, "tcp-") {
+		return "tcp-" + lP.Name
+	}
+	return lP.Name
+}
+
 // GetReplicas returns replicas used by the Istio Ingress deployment
 func (iIConfig *IstioIngressConfig) GetReplicas() int32 {
 	if iIConfig.Replicas == 0 {
diff --git a/api/v1beta1/zz_generated.deepcopy.go b/api/v1beta1/zz_generated.deepcopy.go
index a4ba8005..f777360e 100644
--- a/api/v1beta1/zz_generated.deepcopy.go
+++ b/api/v1beta1/zz_generated.deepcopy.go
@@ -134,6 +134,21 @@ func (in *BrokerState) DeepCopy() *BrokerState {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CommonListenerSpec) DeepCopyInto(out *CommonListenerSpec) {
+	*out = *in
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CommonListenerSpec.
+func (in *CommonListenerSpec) DeepCopy() *CommonListenerSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(CommonListenerSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CruiseControlConfig) DeepCopyInto(out *CruiseControlConfig) {
 	*out = *in
@@ -248,6 +263,7 @@ func (in *EnvoyConfig) DeepCopy() *EnvoyConfig {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ExternalListenerConfig) DeepCopyInto(out *ExternalListenerConfig) {
 	*out = *in
+	out.CommonListenerSpec = in.CommonListenerSpec
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ExternalListenerConfig.
@@ -285,6 +301,7 @@ func (in *GracefulActionState) DeepCopy() *GracefulActionState {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *InternalListenerConfig) DeepCopyInto(out *InternalListenerConfig) {
 	*out = *in
+	out.CommonListenerSpec = in.CommonListenerSpec
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InternalListenerConfig.
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index 99fd262d..645acdf7 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -3,14 +3,18 @@
 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition
 metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.2.5
   creationTimestamp: null
   name: kafkaclusters.kafka.banzaicloud.io
 spec:
   group: kafka.banzaicloud.io
   names:
     kind: KafkaCluster
+    listKind: KafkaClusterList
     plural: kafkaclusters
-  scope: ""
+    singular: kafkacluster
+  scope: Namespaced
   subresources:
     status: {}
   validation:
@@ -291,13 +295,21 @@ spec:
                     properties:
                       limits:
                         additionalProperties:
-                          type: string
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
                         description: 'Limits describes the maximum amount of compute
                           resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                         type: object
                       requests:
                         additionalProperties:
-                          type: string
+                          anyOf:
+                          - type: integer
+                          - type: string
+                          pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                          x-kubernetes-int-or-string: true
                         description: 'Requests describes the minimum amount of compute
                           resources required. If Requests is omitted for a container,
                           it defaults to Limits if that is explicitly specified, otherwise
@@ -359,13 +371,21 @@ spec:
                               properties:
                                 limits:
                                   additionalProperties:
-                                    type: string
+                                    anyOf:
+                                    - type: integer
+                                    - type: string
+                                    pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                    x-kubernetes-int-or-string: true
                                   description: 'Limits describes the maximum amount
                                     of compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                                   type: object
                                 requests:
                                   additionalProperties:
-                                    type: string
+                                    anyOf:
+                                    - type: integer
+                                    - type: string
+                                    pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                    x-kubernetes-int-or-string: true
                                   description: 'Requests describes the minimum amount
                                     of compute resources required. If Requests is
                                     omitted for a container, it defaults to Limits
@@ -735,13 +755,21 @@ spec:
                         properties:
                           limits:
                             additionalProperties:
-                              type: string
+                              anyOf:
+                              - type: integer
+                              - type: string
+                              pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                              x-kubernetes-int-or-string: true
                             description: 'Limits describes the maximum amount of compute
                               resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                             type: object
                           requests:
                             additionalProperties:
-                              type: string
+                              anyOf:
+                              - type: integer
+                              - type: string
+                              pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                              x-kubernetes-int-or-string: true
                             description: 'Requests describes the minimum amount of
                               compute resources required. If Requests is omitted for
                               a container, it defaults to Limits if that is explicitly
@@ -806,13 +834,21 @@ spec:
                                   properties:
                                     limits:
                                       additionalProperties:
-                                        type: string
+                                        anyOf:
+                                        - type: integer
+                                        - type: string
+                                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                        x-kubernetes-int-or-string: true
                                       description: 'Limits describes the maximum amount
                                         of compute resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                                       type: object
                                     requests:
                                       additionalProperties:
-                                        type: string
+                                        anyOf:
+                                        - type: integer
+                                        - type: string
+                                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                                        x-kubernetes-int-or-string: true
                                       description: 'Requests describes the minimum
                                         amount of compute resources required. If Requests
                                         is omitted for a container, it defaults to
@@ -996,13 +1032,21 @@ spec:
                   properties:
                     limits:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Limits describes the maximum amount of compute
                         resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                       type: object
                     requests:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Requests describes the minimum amount of compute
                         resources required. If Requests is omitted for a container,
                         it defaults to Limits if that is explicitly specified, otherwise
@@ -1104,13 +1148,21 @@ spec:
                   properties:
                     limits:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Limits describes the maximum amount of compute
                         resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                       type: object
                     requests:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Requests describes the minimum amount of compute
                         resources required. If Requests is omitted for a container,
                         it defaults to Limits if that is explicitly specified, otherwise
@@ -1162,6 +1214,7 @@ spec:
             headlessServiceEnabled:
               type: boolean
             ingressController:
+              default: envoy
               enum:
               - envoy
               - istioingress
@@ -1269,13 +1322,21 @@ spec:
                   properties:
                     limits:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Limits describes the maximum amount of compute
                         resources allowed. More info: https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/'
                       type: object
                     requests:
                       additionalProperties:
-                        type: string
+                        anyOf:
+                        - type: integer
+                        - type: string
+                        pattern: ^(\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))(([KMGTPE]i)|[numkMGTPE]|([eE](\+|-)?(([0-9]+(\.[0-9]*)?)|(\.[0-9]+))))?$
+                        x-kubernetes-int-or-string: true
                       description: 'Requests describes the minimum amount of compute
                         resources required. If Requests is omitted for a container,
                         it defaults to Limits if that is explicitly specified, otherwise
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkatopics.yaml b/config/base/crds/kafka.banzaicloud.io_kafkatopics.yaml
index 374af6f7..5b110ff6 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkatopics.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkatopics.yaml
@@ -3,14 +3,18 @@
 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition
 metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.2.5
   creationTimestamp: null
   name: kafkatopics.kafka.banzaicloud.io
 spec:
   group: kafka.banzaicloud.io
   names:
     kind: KafkaTopic
+    listKind: KafkaTopicList
     plural: kafkatopics
-  scope: ""
+    singular: kafkatopic
+  scope: Namespaced
   subresources:
     status: {}
   validation:
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml b/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
index 87e20454..7b4063c7 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkausers.yaml
@@ -3,14 +3,18 @@
 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition
 metadata:
+  annotations:
+    controller-gen.kubebuilder.io/version: v0.2.5
   creationTimestamp: null
   name: kafkausers.kafka.banzaicloud.io
 spec:
   group: kafka.banzaicloud.io
   names:
     kind: KafkaUser
+    listKind: KafkaUserList
     plural: kafkausers
-  scope: ""
+    singular: kafkauser
+  scope: Namespaced
   subresources:
     status: {}
   validation:
@@ -43,12 +47,40 @@ spec:
               required:
               - name
               type: object
+            createCert:
+              type: boolean
             dnsNames:
               items:
                 type: string
               type: array
             includeJKS:
               type: boolean
+            pkiBackendSpec:
+              properties:
+                issuerRef:
+                  description: ObjectReference is a reference to an object with a
+                    given name, kind and group.
+                  properties:
+                    group:
+                      type: string
+                    kind:
+                      type: string
+                    name:
+                      type: string
+                  required:
+                  - name
+                  type: object
+                pkiBackend:
+                  description: PKIBackend represents an interface implementing the
+                    PKIManager
+                  enum:
+                  - cert-manager
+                  - vault
+                  type: string
+              required:
+              - issuerRef
+              - pkiBackend
+              type: object
             secretName:
               type: string
             topicGrants:
diff --git a/config/samples/example-user-with-pkibackend.yaml b/config/samples/example-user-with-pkibackend.yaml
new file mode 100644
index 00000000..cb22a802
--- /dev/null
+++ b/config/samples/example-user-with-pkibackend.yaml
@@ -0,0 +1,19 @@
+apiVersion: kafka.banzaicloud.io/v1alpha1
+kind: KafkaUser
+metadata:
+  name: example-kafkauser-with-pki
+  namespace: kafka
+spec:
+  clusterRef:
+    name: kafka
+  secretName: example-kafkauser-secret
+  topicGrants:
+    - topicName: example-topic
+      accessType: read
+    - topicName: example-topic
+      accessType: write
+  pkiBackendSpec:
+    pkiBackend: "cert-manager"
+    issuerRef:
+      name: "ca-issuer"
+      kind: "ClusterIssuer"
diff --git a/controllers/kafkacluster_controller.go b/controllers/kafkacluster_controller.go
index d03ac011..6aa62fdf 100644
--- a/controllers/kafkacluster_controller.go
+++ b/controllers/kafkacluster_controller.go
@@ -250,33 +250,33 @@ func (r *KafkaClusterReconciler) checkFinalizers(ctx context.Context, log logr.L
 		}, nil
 	}
 
-	if cluster.Spec.ListenersConfig.SSLSecrets != nil {
-		// If we haven't deleted all kafkausers yet, iterate namespaces and delete all kafkausers
-		// with the matching label.
-		if util.StringSliceContains(cluster.GetFinalizers(), clusterUsersFinalizer) {
-			log.Info(fmt.Sprintf("Sending delete kafkausers request to all namespaces for cluster %s/%s", cluster.Namespace, cluster.Name))
-			for _, ns := range namespaces {
-				if err := r.Client.DeleteAllOf(
-					ctx,
-					&v1alpha1.KafkaUser{},
-					client.InNamespace(ns),
-					client.MatchingLabels{clusterRefLabel: clusterLabelString(cluster)},
-				); err != nil {
-					if client.IgnoreNotFound(err) != nil {
-						return requeueWithError(log, "failed to send delete request for children kafkausers", err)
-					}
-					log.Info(fmt.Sprintf("No matching kafkausers in namespace: %s", ns))
+	// If we haven't deleted all kafkausers yet, iterate namespaces and delete all kafkausers
+	// with the matching label.
+	if util.StringSliceContains(cluster.GetFinalizers(), clusterUsersFinalizer) {
+		log.Info(fmt.Sprintf("Sending delete kafkausers request to all namespaces for cluster %s/%s", cluster.Namespace, cluster.Name))
+		for _, ns := range namespaces {
+			if err := r.Client.DeleteAllOf(
+				ctx,
+				&v1alpha1.KafkaUser{},
+				client.InNamespace(ns),
+				client.MatchingLabels{clusterRefLabel: clusterLabelString(cluster)},
+			); err != nil {
+				if client.IgnoreNotFound(err) != nil {
+					return requeueWithError(log, "failed to send delete request for children kafkausers", err)
 				}
+				log.Info(fmt.Sprintf("No matching kafkausers in namespace: %s", ns))
 			}
-			if cluster, err = r.removeFinalizer(ctx, cluster, clusterUsersFinalizer); err != nil {
-				return requeueWithError(log, "failed to remove users finalizer from kafkacluster", err)
-			}
 		}
+		if cluster, err = r.removeFinalizer(ctx, cluster, clusterUsersFinalizer); err != nil {
+			return requeueWithError(log, "failed to remove users finalizer from kafkacluster", err)
+		}
+	}
+	if cluster.Spec.ListenersConfig.SSLSecrets != nil {
 
 		// Do any necessary PKI cleanup - a PKI backend should make sure any
 		// user finalizations are done before it does its final cleanup
 		log.Info("Tearing down any PKI resources for the kafkacluster")
-		if err = pki.GetPKIManager(r.Client, cluster).FinalizePKI(ctx, log); err != nil {
+		if err = pki.GetPKIManager(r.Client, cluster, v1beta1.PKIBackendProvided).FinalizePKI(ctx, log); err != nil {
 			switch err.(type) {
 			case errorfactory.ResourceNotReady:
 				log.Info("The PKI is not ready to be torn down")
@@ -313,10 +313,7 @@ func topicListToStrSlice(list v1alpha1.KafkaTopicList) []string {
 }
 
 func (r *KafkaClusterReconciler) ensureFinalizers(ctx context.Context, cluster *v1beta1.KafkaCluster) (updated *v1beta1.KafkaCluster, err error) {
-	finalizers := []string{clusterFinalizer, clusterTopicsFinalizer}
-	if cluster.Spec.ListenersConfig.SSLSecrets != nil {
-		finalizers = append(finalizers, clusterUsersFinalizer)
-	}
+	finalizers := []string{clusterFinalizer, clusterTopicsFinalizer, clusterUsersFinalizer}
 	for _, finalizer := range finalizers {
 		if util.StringSliceContains(cluster.GetFinalizers(), finalizer) {
 			continue
diff --git a/controllers/kafkauser_controller.go b/controllers/kafkauser_controller.go
index baab6819..6d6e37fd 100644
--- a/controllers/kafkauser_controller.go
+++ b/controllers/kafkauser_controller.go
@@ -31,6 +31,8 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
 	"github.com/banzaicloud/kafka-operator/api/v1alpha1"
 	"github.com/banzaicloud/kafka-operator/api/v1beta1"
 	"github.com/banzaicloud/kafka-operator/pkg/errorfactory"
@@ -38,9 +40,6 @@ import (
 	"github.com/banzaicloud/kafka-operator/pkg/pki"
 	"github.com/banzaicloud/kafka-operator/pkg/util"
 	kafkautil "github.com/banzaicloud/kafka-operator/pkg/util/kafka"
-	pkicommon "github.com/banzaicloud/kafka-operator/pkg/util/pki"
-
-	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -135,54 +134,72 @@ func (r *KafkaUserReconciler) Reconcile(request reconcile.Request) (reconcile.Re
 	}
 	// Avoid panic if the user wants to create a kafka user but the cluster is in plaintext mode
 	// TODO: refactor this and use webhook to validate if the cluster is eligible to create a kafka user
-	if cluster.Spec.ListenersConfig.SSLSecrets == nil {
-		return requeueWithError(reqLogger, "could not create kafka user since cluster does not use ssl", errors.New("failed to create kafka user"))
+	if cluster.Spec.ListenersConfig.SSLSecrets == nil && instance.Spec.PKIBackendSpec.PKIBackend == "" {
+		return requeueWithError(reqLogger, "could not create kafka user since user specific PKI not configured", errors.New("failed to create kafka user"))
 	}
 
-	pkiManager := pki.GetPKIManager(r.Client, cluster)
+	var kafkaUser string
 
-	// Reconcile no matter what to get a user certificate instance for ACL management
-	// TODO (tinyzimmer): This can go wrong if the user made a mistake in their secret path
-	// using the vault backend, then tried to delete and fix it. Should probably
-	// have the PKIManager export a GetUserCertificate specifically for deletions
-	// that will allow the error to fall through if the certificate doesn't exist.
-	user, err := pkiManager.ReconcileUserCertificate(ctx, instance, r.Scheme)
-	if err != nil {
-		switch errors.Cause(err).(type) {
-		case errorfactory.ResourceNotReady:
-			reqLogger.Info("generated secret not found, may not be ready")
-			return ctrl.Result{
-				Requeue:      true,
-				RequeueAfter: time.Duration(5) * time.Second,
-			}, nil
-		case errorfactory.FatalReconcileError:
-			// TODO: (tinyzimmer) - Sleep for longer for now to give user time to see the error
-			// But really we should catch these kinds of issues in a pre-admission hook in a future PR
-			// The user can fix while this is looping and it will pick it up next reconcile attempt
-			reqLogger.Error(err, "Fatal error attempting to reconcile the user certificate. If using vault perhaps a permissions issue or improperly configured PKI?")
-			return ctrl.Result{
-				Requeue:      true,
-				RequeueAfter: time.Duration(15) * time.Second,
-			}, nil
-		case errorfactory.VaultAPIFailure:
-			// Same as above in terms of things that could be checked pre-flight on the cluster
-			reqLogger.Error(err, "Vault API error attempting to reconcile the user certificate. If using vault perhaps a permissions issue or improperly configured PKI?")
-			return ctrl.Result{
-				Requeue:      true,
-				RequeueAfter: time.Duration(15) * time.Second,
-			}, nil
-		default:
-			return requeueWithError(reqLogger, "failed to reconcile user secret", err)
+	if instance.Spec.GetIfCertShouldBeCreated() {
+
+		var backend v1beta1.PKIBackend
+		if instance.Spec.PKIBackendSpec.PKIBackend != "" {
+			backend = instance.Spec.PKIBackendSpec.PKIBackend
+		} else {
+			backend = v1beta1.PKIBackendProvided
 		}
+
+		pkiManager := pki.GetPKIManager(r.Client, cluster, backend)
+
+		// Reconcile no matter what to get a user certificate instance for ACL management
+		// TODO (tinyzimmer): This can go wrong if the user made a mistake in their secret path
+		// using the vault backend, then tried to delete and fix it. Should probably
+		// have the PKIManager export a GetUserCertificate specifically for deletions
+		// that will allow the error to fall through if the certificate doesn't exist.
+		user, err := pkiManager.ReconcileUserCertificate(ctx, instance, r.Scheme)
+		if err != nil {
+			switch errors.Cause(err).(type) {
+			case errorfactory.ResourceNotReady:
+				reqLogger.Info("generated secret not found, may not be ready")
+				return ctrl.Result{
+					Requeue:      true,
+					RequeueAfter: time.Duration(5) * time.Second,
+				}, nil
+			case errorfactory.FatalReconcileError:
+				// TODO: (tinyzimmer) - Sleep for longer for now to give user time to see the error
+				// But really we should catch these kinds of issues in a pre-admission hook in a future PR
+				// The user can fix while this is looping and it will pick it up next reconcile attempt
+				reqLogger.Error(err, "Fatal error attempting to reconcile the user certificate. If using vault perhaps a permissions issue or improperly configured PKI?")
+				return ctrl.Result{
+					Requeue:      true,
+					RequeueAfter: time.Duration(15) * time.Second,
+				}, nil
+			case errorfactory.VaultAPIFailure:
+				// Same as above in terms of things that could be checked pre-flight on the cluster
+				reqLogger.Error(err, "Vault API error attempting to reconcile the user certificate. If using vault perhaps a permissions issue or improperly configured PKI?")
+				return ctrl.Result{
+					Requeue:      true,
+					RequeueAfter: time.Duration(15) * time.Second,
+				}, nil
+			default:
+				return requeueWithError(reqLogger, "failed to reconcile user secret", err)
+			}
+		}
+		kafkaUser = user.DN()
+		// check if marked for deletion and remove created certs
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			reqLogger.Info("Kafka user is marked for deletion, revoking certificates")
+			if err = pkiManager.FinalizeUserCertificate(ctx, instance); err != nil {
+				return requeueWithError(reqLogger, "failed to finalize user certificate", err)
+			}
+		}
+	} else {
+		kafkaUser = instance.Name
 	}
 
-	// check if marked for deletion
+	// check if marked for deletion and remove kafka ACLs
 	if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
-		reqLogger.Info("Kafka user is marked for deletion, revoking certificates")
-		if err = pkiManager.FinalizeUserCertificate(ctx, instance); err != nil {
-			return requeueWithError(reqLogger, "failed to finalize user certificate", err)
-		}
-		return r.checkFinalizers(ctx, reqLogger, cluster, instance, user)
+		return r.checkFinalizers(ctx, reqLogger, cluster, instance, kafkaUser)
 	}
 
 	// ensure a kafkaCluster label
@@ -200,9 +217,9 @@ func (r *KafkaUserReconciler) Reconcile(request reconcile.Request) (reconcile.Re
 
 		// TODO (tinyzimmer): Should probably take this opportunity to see if we are removing any ACLs
 		for _, grant := range instance.Spec.TopicGrants {
-			reqLogger.Info(fmt.Sprintf("Ensuring %s ACLs for User: %s -> Topic: %s", grant.AccessType, user.DN(), grant.TopicName))
+			reqLogger.Info(fmt.Sprintf("Ensuring %s ACLs for User: %s -> Topic: %s", grant.AccessType, kafkaUser, grant.TopicName))
 			// CreateUserACLs returns no error if the ACLs already exist
-			if err = broker.CreateUserACLs(grant.AccessType, grant.PatternType, user.DN(), grant.TopicName); err != nil {
+			if err = broker.CreateUserACLs(grant.AccessType, grant.PatternType, kafkaUser, grant.TopicName); err != nil {
 				return requeueWithError(reqLogger, "failed to ensure ACLs for kafkauser", err)
 			}
 		}
@@ -221,7 +238,7 @@ func (r *KafkaUserReconciler) Reconcile(request reconcile.Request) (reconcile.Re
 		State: v1alpha1.UserStateCreated,
 	}
 	if len(instance.Spec.TopicGrants) > 0 {
-		instance.Status.ACLs = kafkautil.GrantsToACLStrings(user.DN(), instance.Spec.TopicGrants)
+		instance.Status.ACLs = kafkautil.GrantsToACLStrings(kafkaUser, instance.Spec.TopicGrants)
 	}
 	if err := r.Client.Status().Update(ctx, instance); err != nil {
 		return requeueWithError(reqLogger, "failed to update kafkauser status", err)
@@ -249,7 +266,7 @@ func (r *KafkaUserReconciler) updateAndFetchLatest(ctx context.Context, user *v1
 	return user, nil
 }
 
-func (r *KafkaUserReconciler) checkFinalizers(ctx context.Context, reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, instance *v1alpha1.KafkaUser, user *pkicommon.UserCertificate) (reconcile.Result, error) {
+func (r *KafkaUserReconciler) checkFinalizers(ctx context.Context, reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, instance *v1alpha1.KafkaUser, user string) (reconcile.Result, error) {
 	// run finalizers
 	var err error
 	if util.StringSliceContains(instance.GetFinalizers(), userFinalizer) {
@@ -272,7 +289,7 @@ func (r *KafkaUserReconciler) removeFinalizer(ctx context.Context, user *v1alpha
 	return err
 }
 
-func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, user *pkicommon.UserCertificate) error {
+func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, cluster *v1beta1.KafkaCluster, user string) error {
 	if k8sutil.IsMarkedForDeletion(cluster.ObjectMeta) {
 		reqLogger.Info("Cluster is being deleted, skipping ACL deletion")
 		return nil
@@ -284,7 +301,7 @@ func (r *KafkaUserReconciler) finalizeKafkaUserACLs(reqLogger logr.Logger, clust
 		return err
 	}
 	defer close()
-	if err = broker.DeleteUserACLs(user.DN()); err != nil {
+	if err = broker.DeleteUserACLs(user); err != nil {
 		return err
 	}
 	return nil
diff --git a/pkg/kafkaclient/config.go b/pkg/kafkaclient/config.go
index e0aa23ad..4ee260c8 100644
--- a/pkg/kafkaclient/config.go
+++ b/pkg/kafkaclient/config.go
@@ -41,7 +41,7 @@ func ClusterConfig(client client.Client, cluster *v1beta1.KafkaCluster) (*KafkaC
 	conf.BrokerURI = generateKafkaAddress(cluster)
 	conf.OperationTimeout = kafkaDefaultTimeout
 	if cluster.Spec.ListenersConfig.SSLSecrets != nil && useSSL(cluster) {
-		tlsConfig, err := pki.GetPKIManager(client, cluster).GetControllerTLSConfig()
+		tlsConfig, err := pki.GetPKIManager(client, cluster, v1beta1.PKIBackendProvided).GetControllerTLSConfig()
 		if err != nil {
 			return conf, err
 		}
diff --git a/pkg/pki/certmanagerpki/certmanager_user.go b/pkg/pki/certmanagerpki/certmanager_user.go
index 1de24b29..134c5de7 100644
--- a/pkg/pki/certmanagerpki/certmanager_user.go
+++ b/pkg/pki/certmanagerpki/certmanager_user.go
@@ -138,7 +138,7 @@ func (c *certManager) getUserSecret(ctx context.Context, user *v1alpha1.KafkaUse
 
 // clusterCertificateForUser generates a Certificate object for a KafkaUser
 func (c *certManager) clusterCertificateForUser(user *v1alpha1.KafkaUser, scheme *runtime.Scheme) *certv1.Certificate {
-	caName, caKind := c.getCA()
+	caName, caKind := c.getCA(user)
 	cert := &certv1.Certificate{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      user.Name,
@@ -163,8 +163,13 @@ func (c *certManager) clusterCertificateForUser(user *v1alpha1.KafkaUser, scheme
 }
 
 // getCA returns the CA name/kind for the KafkaCluster
-func (c *certManager) getCA() (caName, caKind string) {
-	issuerRef := c.cluster.Spec.ListenersConfig.SSLSecrets.IssuerRef
+func (c *certManager) getCA(user *v1alpha1.KafkaUser) (caName, caKind string) {
+	var issuerRef *certmeta.ObjectReference
+	if user.Spec.PKIBackendSpec.IssuerRef != nil {
+		issuerRef = user.Spec.PKIBackendSpec.IssuerRef
+	} else {
+		issuerRef = c.cluster.Spec.ListenersConfig.SSLSecrets.IssuerRef
+	}
 	if issuerRef != nil {
 		caName = issuerRef.Name
 		caKind = issuerRef.Kind
diff --git a/pkg/pki/pki_manager.go b/pkg/pki/pki_manager.go
index 0a40c256..0891d16c 100644
--- a/pkg/pki/pki_manager.go
+++ b/pkg/pki/pki_manager.go
@@ -32,8 +32,14 @@ import (
 var MockBackend = v1beta1.PKIBackend("mock")
 
 // GetPKIManager returns a PKI/User manager interface for a given cluster
-func GetPKIManager(client client.Client, cluster *v1beta1.KafkaCluster) pki.Manager {
-	switch cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend {
+func GetPKIManager(client client.Client, cluster *v1beta1.KafkaCluster, pkiBackend v1beta1.PKIBackend) pki.Manager {
+	var backend v1beta1.PKIBackend
+	if pkiBackend == v1beta1.PKIBackendProvided {
+		backend = cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend
+	} else {
+		backend = pkiBackend
+	}
+	switch backend {
 
 	// Use cert-manager for pki backend
 	case v1beta1.PKIBackendCertManager:
diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go
index d1aaa278..9a55e211 100644
--- a/pkg/resources/cruisecontrol/configmap.go
+++ b/pkg/resources/cruisecontrol/configmap.go
@@ -148,7 +148,7 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger)
 	if err != nil {
 		log.Error(err, "Could not marshal cruise control capacity config")
 	}
-	log.Info("Generated capacity config was successful with values:", result)
+	log.Info(fmt.Sprintf("Generated capacity config was successful with values: %s", result))
 
 	return string(result)
 }
diff --git a/pkg/resources/kafka/allBrokerService.go b/pkg/resources/kafka/allBrokerService.go
index 696f5b45..8b28b315 100644
--- a/pkg/resources/kafka/allBrokerService.go
+++ b/pkg/resources/kafka/allBrokerService.go
@@ -31,7 +31,7 @@ func (r *Reconciler) allBrokerService() runtime.Object {
 
 	for _, iListeners := range r.KafkaCluster.Spec.ListenersConfig.InternalListeners {
 		usedPorts = append(usedPorts, corev1.ServicePort{
-			Name:       strings.ReplaceAll(iListeners.Name, "_", ""),
+			Name:       strings.ReplaceAll(iListeners.GetListenerServiceName(), "_", ""),
 			Port:       iListeners.ContainerPort,
 			TargetPort: intstr.FromInt(int(iListeners.ContainerPort)),
 			Protocol:   corev1.ProtocolTCP,
@@ -40,7 +40,7 @@ func (r *Reconciler) allBrokerService() runtime.Object {
 	//Append external listener ports as well to allow using this service for metadata fetch
 	for _, eListeners := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
 		usedPorts = append(usedPorts, corev1.ServicePort{
-			Name:     strings.ReplaceAll(eListeners.Name, "_", ""),
+			Name:     strings.ReplaceAll(eListeners.GetListenerServiceName(), "_", ""),
 			Port:     eListeners.ContainerPort,
 			Protocol: corev1.ProtocolTCP,
 		})
diff --git a/pkg/resources/kafka/headlessService.go b/pkg/resources/kafka/headlessService.go
index e89e6da5..0ad693c7 100644
--- a/pkg/resources/kafka/headlessService.go
+++ b/pkg/resources/kafka/headlessService.go
@@ -32,7 +32,7 @@ func (r *Reconciler) headlessService() runtime.Object {
 
 	for _, iListeners := range r.KafkaCluster.Spec.ListenersConfig.InternalListeners {
 		usedPorts = append(usedPorts, corev1.ServicePort{
-			Name:       strings.ReplaceAll(iListeners.Name, "_", ""),
+			Name:       strings.ReplaceAll(iListeners.GetListenerServiceName(), "_", ""),
 			Port:       iListeners.ContainerPort,
 			TargetPort: intstr.FromInt(int(iListeners.ContainerPort)),
 			Protocol:   corev1.ProtocolTCP,
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index 3c0650f4..78940c0d 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -210,7 +210,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 	// Setup the PKI if using SSL
 	if r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil {
 		// reconcile the PKI
-		if err := pki.GetPKIManager(r.Client, r.KafkaCluster).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, lbIPs); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/resources/kafka/pod.go b/pkg/resources/kafka/pod.go
index 22bdef07..302aca96 100644
--- a/pkg/resources/kafka/pod.go
+++ b/pkg/resources/kafka/pod.go
@@ -37,7 +37,7 @@ func (r *Reconciler) pod(id int32, brokerConfig *v1beta1.BrokerConfig, pvcs []co
 
 	for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
 		kafkaBrokerContainerPorts = append(kafkaBrokerContainerPorts, corev1.ContainerPort{
-			Name:          strings.ReplaceAll(eListener.Name, "_", "-"),
+			Name:          strings.ReplaceAll(eListener.GetListenerServiceName(), "_", "-"),
 			ContainerPort: eListener.ContainerPort,
 			Protocol:      corev1.ProtocolTCP,
 		})
@@ -45,7 +45,7 @@ func (r *Reconciler) pod(id int32, brokerConfig *v1beta1.BrokerConfig, pvcs []co
 
 	for _, iListener := range r.KafkaCluster.Spec.ListenersConfig.InternalListeners {
 		kafkaBrokerContainerPorts = append(kafkaBrokerContainerPorts, corev1.ContainerPort{
-			Name:          strings.ReplaceAll(iListener.Name, "_", "-"),
+			Name:          strings.ReplaceAll(iListener.GetListenerServiceName(), "_", "-"),
 			ContainerPort: iListener.ContainerPort,
 			Protocol:      corev1.ProtocolTCP,
 		})
diff --git a/pkg/resources/kafka/service.go b/pkg/resources/kafka/service.go
index 96b2bd23..c72cf551 100644
--- a/pkg/resources/kafka/service.go
+++ b/pkg/resources/kafka/service.go
@@ -33,7 +33,7 @@ func (r *Reconciler) service(id int32, log logr.Logger) runtime.Object {
 
 	for _, iListeners := range r.KafkaCluster.Spec.ListenersConfig.InternalListeners {
 		usedPorts = append(usedPorts, corev1.ServicePort{
-			Name:       strings.ReplaceAll(iListeners.Name, "_", ""),
+			Name:       strings.ReplaceAll(iListeners.GetListenerServiceName(), "_", ""),
 			Port:       iListeners.ContainerPort,
 			TargetPort: intstr.FromInt(int(iListeners.ContainerPort)),
 			Protocol:   corev1.ProtocolTCP,
@@ -42,7 +42,7 @@ func (r *Reconciler) service(id int32, log logr.Logger) runtime.Object {
 	if r.KafkaCluster.Spec.ListenersConfig.ExternalListeners != nil {
 		for _, eListener := range r.KafkaCluster.Spec.ListenersConfig.ExternalListeners {
 			usedPorts = append(usedPorts, corev1.ServicePort{
-				Name:       eListener.Name,
+				Name:       eListener.GetListenerServiceName(),
 				Protocol:   corev1.ProtocolTCP,
 				Port:       eListener.ContainerPort,
 				TargetPort: intstr.FromInt(int(eListener.ContainerPort)),


Test patch:
diff --git a/controllers/controller_common_test.go b/controllers/controller_common_test.go
index f103f4dd..825af307 100644
--- a/controllers/controller_common_test.go
+++ b/controllers/controller_common_test.go
@@ -78,7 +78,10 @@ func TestNewBrokerConnection(t *testing.T) {
 	cluster.Spec = v1beta1.KafkaClusterSpec{
 		ListenersConfig: v1beta1.ListenersConfig{
 			InternalListeners: []v1beta1.InternalListenerConfig{
-				{ContainerPort: 9092},
+				{CommonListenerSpec: v1beta1.CommonListenerSpec{
+					ContainerPort: 9092,
+				},
+				},
 			},
 		},
 	}
diff --git a/internal/alertmanager/currentalert/current_alerts_test.go b/internal/alertmanager/currentalert/current_alerts_test.go
index f158fa32..faa4ee94 100644
--- a/internal/alertmanager/currentalert/current_alerts_test.go
+++ b/internal/alertmanager/currentalert/current_alerts_test.go
@@ -103,11 +103,11 @@ func TestGetCurrentAlerts(t *testing.T) {
 			HeadlessServiceEnabled: true,
 			ListenersConfig: v1beta1.ListenersConfig{
 				InternalListeners: []v1beta1.InternalListenerConfig{
-					{
-						Type:                            "plaintext",
-						Name:                            "planitext",
+					{CommonListenerSpec: v1beta1.CommonListenerSpec{
+						Type:          "plaintext",
+						Name:          "planitext",
+						ContainerPort: 29092},
 						UsedForInnerBrokerCommunication: true,
-						ContainerPort:                   29092,
 					},
 				},
 			},
diff --git a/pkg/kafkaclient/config_test.go b/pkg/kafkaclient/config_test.go
index 3383381a..d3f0f466 100644
--- a/pkg/kafkaclient/config_test.go
+++ b/pkg/kafkaclient/config_test.go
@@ -32,8 +32,9 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	cluster.Namespace = "test"
 	cluster.Spec = v1beta1.KafkaClusterSpec{}
 	cluster.Spec.ListenersConfig = v1beta1.ListenersConfig{}
-	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{
-		v1beta1.InternalListenerConfig{ContainerPort: 80},
+	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{{
+		CommonListenerSpec: v1beta1.CommonListenerSpec{ContainerPort: 80},
+	},
 	}
 	cluster.Spec.ListenersConfig.SSLSecrets = &v1beta1.SSLSecrets{
 		PKIBackend: pki.MockBackend,
diff --git a/pkg/pki/certmanagerpki/certmanager_test.go b/pkg/pki/certmanagerpki/certmanager_test.go
index bf96c5ff..b0390955 100644
--- a/pkg/pki/certmanagerpki/certmanager_test.go
+++ b/pkg/pki/certmanagerpki/certmanager_test.go
@@ -37,7 +37,9 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	cluster.Spec = v1beta1.KafkaClusterSpec{}
 	cluster.Spec.ListenersConfig = v1beta1.ListenersConfig{}
 	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{
-		{ContainerPort: 9092},
+		{CommonListenerSpec: v1beta1.CommonListenerSpec{
+			ContainerPort: 9092,
+		}},
 	}
 	cluster.Spec.ListenersConfig.SSLSecrets = &v1beta1.SSLSecrets{
 		TLSSecretName:   "test-controller",
diff --git a/pkg/pki/pki_manager_test.go b/pkg/pki/pki_manager_test.go
index 0b472757..c3fb395a 100644
--- a/pkg/pki/pki_manager_test.go
+++ b/pkg/pki/pki_manager_test.go
@@ -39,7 +39,9 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	cluster.Spec = v1beta1.KafkaClusterSpec{}
 	cluster.Spec.ListenersConfig = v1beta1.ListenersConfig{}
 	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{
-		{ContainerPort: 80},
+		{CommonListenerSpec: v1beta1.CommonListenerSpec{
+			ContainerPort: 80,
+		}},
 	}
 	cluster.Spec.ListenersConfig.SSLSecrets = &v1beta1.SSLSecrets{
 		PKIBackend: MockBackend,
@@ -49,7 +51,7 @@ func newMockCluster() *v1beta1.KafkaCluster {
 
 func TestGetPKIManager(t *testing.T) {
 	cluster := newMockCluster()
-	mock := GetPKIManager(&mockClient{}, cluster)
+	mock := GetPKIManager(&mockClient{}, cluster, v1beta1.PKIBackendProvided)
 	if reflect.TypeOf(mock) != reflect.TypeOf(&mockPKIManager{}) {
 		t.Error("Expected mock client got:", reflect.TypeOf(mock))
 	}
@@ -79,7 +81,7 @@ func TestGetPKIManager(t *testing.T) {
 
 	// Test other getters
 	cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend = v1beta1.PKIBackendCertManager
-	certmanager := GetPKIManager(&mockClient{}, cluster)
+	certmanager := GetPKIManager(&mockClient{}, cluster, v1beta1.PKIBackendProvided)
 	pkiType := reflect.TypeOf(certmanager).String()
 	expected := "*certmanagerpki.certManager"
 	if pkiType != expected {
@@ -88,7 +90,7 @@ func TestGetPKIManager(t *testing.T) {
 
 	// Default should be cert-manager also
 	cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend = v1beta1.PKIBackend("")
-	certmanager = GetPKIManager(&mockClient{}, cluster)
+	certmanager = GetPKIManager(&mockClient{}, cluster, v1beta1.PKIBackendProvided)
 	pkiType = reflect.TypeOf(certmanager).String()
 	expected = "*certmanagerpki.certManager"
 	if pkiType != expected {
@@ -96,7 +98,7 @@ func TestGetPKIManager(t *testing.T) {
 	}
 
 	cluster.Spec.ListenersConfig.SSLSecrets.PKIBackend = v1beta1.PKIBackendVault
-	certmanager = GetPKIManager(&mockClient{}, cluster)
+	certmanager = GetPKIManager(&mockClient{}, cluster, v1beta1.PKIBackendProvided)
 	pkiType = reflect.TypeOf(certmanager).String()
 	expected = "*vaultpki.vaultPKI"
 	if pkiType != expected {
diff --git a/pkg/pki/vaultpki/vault_test.go b/pkg/pki/vaultpki/vault_test.go
index 0c4d8a5b..ce374f31 100644
--- a/pkg/pki/vaultpki/vault_test.go
+++ b/pkg/pki/vaultpki/vault_test.go
@@ -47,7 +47,9 @@ func newMockCluster() *v1beta1.KafkaCluster {
 	cluster.Spec = v1beta1.KafkaClusterSpec{}
 	cluster.Spec.ListenersConfig = v1beta1.ListenersConfig{}
 	cluster.Spec.ListenersConfig.InternalListeners = []v1beta1.InternalListenerConfig{
-		{ContainerPort: 9092},
+		{CommonListenerSpec: v1beta1.CommonListenerSpec{
+			ContainerPort: 9092,
+		}},
 	}
 	cluster.Spec.ListenersConfig.SSLSecrets = &v1beta1.SSLSecrets{
 		TLSSecretName:   "test-controller",
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 05ebb178..ae66971f 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -181,11 +181,12 @@ zookeeper.connect=example.zk:2181/`,
 							ZKPath:      test.zkPath,
 							ListenersConfig: v1beta1.ListenersConfig{
 								InternalListeners: []v1beta1.InternalListenerConfig{
-									{
-										Type:                            "plaintext",
-										Name:                            "internal",
+									{CommonListenerSpec: v1beta1.CommonListenerSpec{
+										Type:          "plaintext",
+										Name:          "internal",
+										ContainerPort: 9092,
+									},
 										UsedForInnerBrokerCommunication: true,
-										ContainerPort:                   9092,
 									},
 								},
 							},
diff --git a/pkg/util/util_test.go b/pkg/util/util_test.go
index ccf9ac23..efe26c87 100644
--- a/pkg/util/util_test.go
+++ b/pkg/util/util_test.go
@@ -207,13 +207,17 @@ func TestConvertStringToInt32(t *testing.T) {
 
 func TestIsSSLEnabledForInternalCommunication(t *testing.T) {
 	lconfig := []v1beta1.InternalListenerConfig{
-		{Type: "ssl"},
+		{
+			CommonListenerSpec: v1beta1.CommonListenerSpec{Type: "ssl"},
+		},
 	}
 	if !IsSSLEnabledForInternalCommunication(lconfig) {
 		t.Error("Expected ssl enabled for internal communication, got disabled")
 	}
 	lconfig = []v1beta1.InternalListenerConfig{
-		{Type: "plaintext"},
+		{
+			CommonListenerSpec: v1beta1.CommonListenerSpec{Type: "plaintext"},
+		},
 	}
 	if IsSSLEnabledForInternalCommunication(lconfig) {
 		t.Error("Expected ssl disabled for internal communication, got enabled")