Configures listeners to connect to Kafka brokers within and outside Kubernetes.

You configure the listeners in the `Kafka` resource.

.Example `Kafka` resource showing listener configuration
[source,yaml,subs="+attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    #...
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
      - name: external2
        port: 9095
        type: ingress
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
            host: bootstrap.myingress.com
          brokers:
          - broker: 0
            host: broker-0.myingress.com
          - broker: 1
            host: broker-1.myingress.com
          - broker: 2
            host: broker-2.myingress.com
    #...
----

[id='property-generic-listeners-{context}']
= `listeners`
You configure Kafka broker listeners using the `listeners` property in the `Kafka` resource.
Listeners are defined as an array.

.Example listener configuration
[source,yaml,subs="+attributes"]
----
listeners:
  - name: plain
    port: 9092
    type: internal
    tls: false
----

The name and port must be unique within the Kafka cluster.
The name can be up to 25 characters long, comprising lower-case letters and numbers.
Allowed port numbers are 9092 and higher with the exception of ports 9404 and 9999, which are already used for Prometheus and JMX.

By specifying a unique name and port for each listener,
you can configure multiple listeners.

= `type`

The type is set as `internal`,
or for external listeners, as `route`, `loadbalancer`, `nodeport`, `ingress` or `cluster-ip`.
You can also configure a `cluster-ip` listener, a type of internal listener you can use to build custom access mechanisms.

internal:: You can configure internal listeners with or without encryption using the `tls` property.
+
.Example `internal` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      #...
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
    #...
----
route:: Configures an external listener to expose Kafka using OpenShift `Routes` and the HAProxy router.
+
A dedicated `Route` is created for every Kafka broker pod.
An additional `Route` is created to serve as a Kafka bootstrap address.
Kafka clients can use these `Routes` to connect to Kafka on port 443.
The client connects on port 443, the default router port, but traffic is then routed to the port you configure,
which is `9094` in this example.
+
.Example `route` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      #...
      - name: external1
        port: 9094
        type: route
        tls: true
    #...
----
ingress:: Configures an external listener to expose Kafka using Kubernetes `Ingress` and the {NginxIngressController}.
+
A dedicated `Ingress` resource is created for every Kafka broker pod.
An additional `Ingress` resource is created to serve as a Kafka bootstrap address.
Kafka clients can use these `Ingress` resources to connect to Kafka on port 443.
The client connects on port 443, the default controller port, but traffic is then routed to the port you configure,
which is `9095` in the following example.
+
You must specify the hostnames used by the bootstrap and per-broker services
using xref:type-GenericKafkaListenerConfigurationBootstrap-reference[`GenericKafkaListenerConfigurationBootstrap`]
and xref:type-GenericKafkaListenerConfigurationBroker-reference[`GenericKafkaListenerConfigurationBroker`] properties.
+
.Example `ingress` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      #...
      - name: external2
        port: 9095
        type: ingress
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
            host: bootstrap.myingress.com
          brokers:
          - broker: 0
            host: broker-0.myingress.com
          - broker: 1
            host: broker-1.myingress.com
          - broker: 2
            host: broker-2.myingress.com
  #...
----
+
NOTE: External listeners using `Ingress` are currently only tested with the {NginxIngressController}.

loadbalancer:: Configures an external listener to expose Kafka using a `Loadbalancer` type `Service`.
+
A new loadbalancer service is created for every Kafka broker pod.
An additional loadbalancer is created to serve as a Kafka _bootstrap_ address.
Loadbalancers listen to the specified port number, which is port `9094` in the following example.
+
You can use the `loadBalancerSourceRanges` property to configure xref:property-listener-config-source-ranges-reference[source ranges] to restrict access to the specified IP addresses.
+
.Example `loadbalancer` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      - name: external3
        port: 9094
        type: loadbalancer
        tls: true
        configuration:
          loadBalancerSourceRanges:
            - 10.0.0.0/8
            - 88.208.76.87/32
    #...
----

nodeport:: Configures an external listener to expose Kafka using a `NodePort` type `Service`.
+
Kafka clients connect directly to the nodes of Kubernetes.
An additional `NodePort` type of service is created to serve as a Kafka bootstrap address.
+
When configuring the advertised addresses for the Kafka broker pods, Strimzi uses the address of the node on which the given pod is running.
You can use `preferredNodePortAddressType` property to configure the xref:property-listener-config-preferredNodePortAddressType-{context}[first address type checked as the node address].
+
.Example `nodeport` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      #...
      - name: external4
        port: 9095
        type: nodeport
        tls: false
        configuration:
          preferredNodePortAddressType: InternalDNS
    #...
----
NOTE: TLS hostname verification is not currently supported when exposing Kafka clusters using node ports.

cluster-ip:: Configures an internal listener to expose Kafka using a per-broker `ClusterIP` type `Service`.
+
The listener does not use a headless service and its DNS names to route traffic to Kafka brokers.
You can use this type of listener to expose a Kafka cluster when using the headless service is unsuitable. 
You might use it with a custom access mechanism, such as one that uses a specific Ingress controller or the Kubernetes Gateway API. 
+
A new `ClusterIP` service is created for each Kafka broker pod.
The service is assigned a `ClusterIP` address to serve as a Kafka _bootstrap_ address with a per-broker port number.
For example, you can configure the listener to expose a Kafka cluster over an Nginx Ingress Controller with TCP port configuration. 
+
.Example `cluster-ip` listener configuration
[source,yaml,subs="+attributes"]
----
#...
spec:
  kafka:
    #...
    listeners:
      - name: external-cluster-ip
        type: cluster-ip
        tls: false
        port: 9096
    #...

----

[id='property-listener-port-{context}']
= `port`

The port number is the port used in the Kafka cluster, which might not be the same port used for access by a client.

* `loadbalancer` listeners use the specified port number, as do `internal` and `cluster-ip` listeners
* `ingress` and `route` listeners use port 443 for access
* `nodeport` listeners use the port number assigned by Kubernetes

For client connection, use the address and port for the bootstrap service of the listener.
You can retrieve this from the status of the `Kafka` resource.

.Example command to retrieve the address and port for client connection
[source,shell,subs=+quotes]
kubectl get kafka _<kafka_cluster_name>_ -o=jsonpath='{.status.listeners[?(@.name=="_<listener_name>_")].bootstrapServers}{"\n"}'

NOTE: Listeners cannot be configured to use the ports set aside for interbroker communication (9090 and 9091) and metrics (9404).

[id='property-listener-tls-{context}']
= `tls`

The TLS property is required.

By default, TLS encryption is not enabled.
To enable it, set the `tls` property to `true`.

For `route` and `ingress` type listeners, TLS encryption must be enabled.

= `authentication`

Authentication for the listener can be specified as:

* mTLS (`tls`)
* SCRAM-SHA-512 (`scram-sha-512`)
* Token-based OAuth 2.0 (`oauth`)
* xref:type-KafkaListenerAuthenticationCustom-reference[Custom (`custom`)]

[id='configuration-listener-network-policy-{context}']
= `networkPolicyPeers`

Use `networkPolicyPeers` to configure network policies that restrict access to a listener at the network level.
The following example shows a `networkPolicyPeers` configuration for a `plain` and a `tls` listener.

In the following example:

* Only application pods matching the labels `app: kafka-sasl-consumer` and `app: kafka-sasl-producer` can connect to the `plain` listener.
The application pods must be running in the same namespace as the Kafka broker.
* Only application pods running in namespaces matching the labels `project: myproject` and `project: myproject2` can connect to the `tls` listener.

The syntax of the `networkPolicyPeers` property is the same as the `from` property in `NetworkPolicy` resources.

.Example network policy configuration
[source,yaml,subs="attributes+"]
----
listeners:
  #...
  - name: plain
    port: 9092
    type: internal
    tls: true
    authentication:
      type: scram-sha-512
    networkPolicyPeers:
      - podSelector:
          matchLabels:
            app: kafka-sasl-consumer
      - podSelector:
          matchLabels:
            app: kafka-sasl-producer
  - name: tls
    port: 9093
    type: internal
    tls: true
    authentication:
      type: tls
    networkPolicyPeers:
      - namespaceSelector:
          matchLabels:
            project: myproject
      - namespaceSelector:
          matchLabels:
            project: myproject2
# ...
----


