operator v1: create internal kafka api port

src/go/k8s/internal/controller/redpanda/cluster_controller.go

@@ -1070,6 +1070,11 @@ func collectClusterPorts(
 		port := redpandaCluster.Spec.Configuration.SchemaRegistry.Port
 		clusterPorts = append(clusterPorts, resources.NamedServicePort{Name: resources.SchemaRegistryPortName, Port: port})
 	}
+	if redpandaPorts.KafkaAPI.Internal != nil {
+		port := redpandaPorts.KafkaAPI.Internal.Port
+		clusterPorts = append(clusterPorts, resources.NamedServicePort{Name: resources.InternalListenerName, Port: port})
+	}
+
 	return clusterPorts
 }
 

src/go/k8s/tests/e2e/kafka-api-cluster-service-internal/00-assert.yaml

@@ -0,0 +1,34 @@
+apiVersion: redpanda.vectorized.io/v1alpha1
+kind: Cluster
+metadata:
+  name: kafka-api-cluster-service-internal
+status:
+  replicas: 1
+  restarting: false
+  conditions:
+    - type: ClusterConfigured
+      status: "True"
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: kafka-api-cluster-service-internal
+status:
+  readyReplicas: 1
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: kafka-api-cluster-service-internal-cluster
+spec:
+  ports:
+    - name: kafka
+      port: 9092
+      protocol: TCP
+      targetPort: 9092
+  type: ClusterIP
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+collectors:
+  - command: ../../../hack/get-redpanda-info.sh

src/go/k8s/tests/e2e/kafka-api-cluster-service-internal/00-create.yaml

@@ -0,0 +1,28 @@
+apiVersion: redpanda.vectorized.io/v1alpha1
+kind: Cluster
+metadata:
+  name: kafka-api-cluster-service-internal
+spec:
+  image: "localhost/redpanda"
+  version: "dev"
+  replicas: 1
+  resources:
+    requests:
+      cpu: 1
+      memory: 1Gi
+    limits:
+      cpu: 1
+      memory: 1Gi
+  configuration:
+    rpcServer:
+      port: 33145
+    kafkaApi:
+      - port: 9092
+        tls:
+          enabled: false
+    adminApi:
+      - port: 9644
+    developerMode: true
+    additionalCommandlineArguments:
+      dump-memory-diagnostics-on-alloc-failure-kind: all
+      abort-on-seastar-bad-alloc: ''

src/go/k8s/tests/e2e/kafka-api-cluster-service-internal/01-assert.yaml

@@ -0,0 +1,4 @@
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+collectors:
+  - command: ../../../hack/get-redpanda-info.sh

src/go/k8s/tests/e2e/kafka-api-cluster-service-internal/01-clean.yaml

@@ -0,0 +1,11 @@
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+delete:
+  - apiVersion: redpanda.vectorized.io/v1alpha1
+    kind: Cluster
+    name: kafka-api-cluster-service-internal
+    namespace: redpanda-system
+  - apiVersion: v1
+    kind: PersistentVolumeClaim
+    name: datadir-kafka-api-cluster-service-internal-0
+    namespace: redpanda-system