// Module included in the following assemblies:
//
// assembly-configuring-kafka-listeners.adoc

[id='con-kafka-listeners-{context}']
= Kafka listeners

You can configure Kafka broker listeners using the `listeners` property in the `Kafka.spec.kafka` resource.
The `listeners` property contains three sub-properties:

* `plain`
* `tls`
* `external`

When none of these properties are defined, the listener will be disabled.

.An example of `listeners` property with all listeners enabled
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  plain: {}
  tls: {}
# ...
----

.An example of `listeners` property with only the plain listener enabled
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  plain: {}
# ...
----

== External listener

The external listener is used to connect to a Kafka cluster from outside of an {ProductPlatformName} environment.
{ProductName} supports three types of external listeners:

* `route`
* `loadbalancer`
* `nodeport`

.Exposing Kafka using {OpenShiftName} Routes

An external listener of type `route` exposes Kafka by using {OpenShiftName} `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.

ifdef::Kubernetes[]
NOTE: `Routes` are available only on {OpenShiftName}. External listeners of type `route` cannot be used on {KubernetesName}.
endif::Kubernetes[]

When exposing Kafka using {OpenShiftName} `Routes`, TLS encryption is always used.

By default, the route hosts are automatically assigned by {OpenShiftName}.
However, you can override the assigned route hosts by specifying the requested hosts in the `overrides` property.
{ProductName} will not perform any validation that the requested hosts are available; you must ensure that they are free and can be used.

.Example of an external listener of type `routes` configured with overrides for {OpenShiftName} route hosts
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  external:
    type: route
    authentication:
      type: tls
    overrides:
      bootstrap:
        host: bootstrap.myrouter.com
      brokers:
      - broker: 0
        host: broker-0.myrouter.com
      - broker: 1
        host: broker-1.myrouter.com
      - broker: 2
        host: broker-2.myrouter.com
# ...
----

For more information on using `Routes` to access Kafka, see xref:proc-accessing-kafka-using-routes-{context}[].

.Exposing Kafka using loadbalancers

External listeners of type `loadbalancer` expose Kafka by using `Loadbalancer` type `Services`.
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 connections on port 9094.

By default, TLS encryption is enabled.
To disable it, set the `tls` field to `false`.

For more information on using loadbalancers to access Kafka, see xref:proc-accessing-kafka-using-loadbalancers-{context}[].

.Exposing Kafka using node ports

External listeners of type `nodeport` expose Kafka by using `NodePort` type `Services`.
When exposing Kafka in this way, Kafka clients connect directly to the nodes of {ProductPlatformName}.
You must enable access to the ports on the {ProductPlatformName} nodes for each client (for example, in firewalls or security groups).
Each Kafka broker pod is then accessible on a separate port.
Additional `NodePort` type `Service` is created to serve as a Kafka bootstrap address.

When configuring the advertised addresses for the Kafka broker pods, {ProductName} uses the address of the node on which the given pod is running.
When selecting the node address, the different address types are used with the following priority:

. ExternalDNS
. ExternalIP
. Hostname
. InternalDNS
. InternalIP

By default, TLS encryption is enabled.
To disable it, set the `tls` field to `false`.

NOTE: TLS hostname verification is not currently supported when exposing Kafka clusters using node ports.

By default, the port numbers used for the bootstrap and broker services are automatically assigned by {ProductPlatformName}.
However, you can override the assigned node ports by specifying the requested port numbers in the `overrides` property.
{ProductName} does not perform any validation on the requested ports; you must ensure that they are free and available for use.

.Example of an external listener configured with overrides for node ports
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  external:
    type: nodeport
    tls: true
    authentication:
      type: tls
    overrides:
      bootstrap:
        nodePort: 32100
      brokers:
      - broker: 0
        nodePort: 32000
      - broker: 1
        nodePort: 32001
      - broker: 2
        nodePort: 32002
# ...
----

For more information on using node ports to access Kafka, see xref:proc-accessing-kafka-using-nodeports-{context}[].

.Customizing advertised addresses on external listeners

By default, {ProductName} tries to automatically determine the hostnames and ports that your Kafka cluster advertises to its clients.
This is not sufficient in all situations, because the infrastructure on which {ProductName} is running might not provide the right hostname or port through which Kafka can be accessed.
You can customize the advertised hostname and port in the `overrides` property of the external listener.
{ProductName} will then automatically configure the advertised address in the Kafka brokers and add it to the broker certificates so it can be used for TLS hostname verification.
Overriding the advertised host and ports is available for all types of external listeners.

.Example of an external listener configured with overrides for advertised addresses
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  external:
    type: route
    authentication:
      type: tls
    overrides:
      brokers:
      - broker: 0
        advertisedHost: example.hostname.0
        advertisedPort: 12340
      - broker: 1
        advertisedHost: example.hostname.1
        advertisedPort: 12341
      - broker: 2
        advertisedHost: example.hostname.2
        advertisedPort: 12342
# ...
----

Additionally, you can specify the name of the bootstrap service.
This name will be added to the broker certificates and can be used for TLS hostname verification.
Adding the additional bootstrap address is available for all types of external listeners.

.Example of an external listener configured with an additional bootstrap address
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  external:
    type: route
    authentication:
      type: tls
    overrides:
      bootstrap:
        address: example.hostname
# ...
----

== Listener authentication

The listener sub-properties can also contain additional configuration.
Both listeners support the `authentication` property. This is used to specify an authentication mechanism specific to that listener:

* mutual TLS authentication (only on the listeners with TLS encryption)
* SCRAM-SHA authentication

If no `authentication` property is specified then the listener does not authenticate clients which connect though that listener.

.An example where the plain listener is configured for SCRAM-SHA authentication and the `tls` listener with mutual TLS authentication
[source,yaml,subs="attributes+"]
----
# ...
listeners:
  plain:
    authentication:
      type: scram-sha-512
  tls:
    authentication:
      type: tls
  external:
    type: loadbalancer
    tls: true
    authentication:
      type: tls
# ...
----


Authentication must be configured when using the User Operator to manage `KafkaUsers`.

== Network policies

{ProductName} automatically creates a `NetworkPolicy` resource for every listener that is enabled on a Kafka broker.
By default, a `NetworkPolicy` grants access to a listener to all applications and namespaces.
If you want to restrict access to a listener to only selected applications or namespaces, use the `networkPolicyPeers` field.
Each listener can have a different `networkPolicyPeers` configuration.

The following example shows a `networkPolicyPeers` configuration for a `plain` and a `tls` listener:

[source,yaml,subs="attributes+"]
----
# ...
listeners:
      plain:
        authentication:
          type: scram-sha-512
        networkPolicyPeers:
          - podSelector:
              matchLabels:
                app: kafka-sasl-consumer
          - podSelector:
              matchLabels:
                app: kafka-sasl-producer
      tls:
        authentication:
          type: tls
        networkPolicyPeers:
          - namespaceSelector:
              matchLabels:
                project: myproject
          - namespaceSelector:
              matchLabels:
                project: myproject2
# ...
----

In the above 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` field is the same as the `from` field in the `NetworkPolicy` resource in {KubernetesName}.
For more information about the schema, see {K8sNetworkPolicyPeerAPI} and the xref:type-KafkaListeners-reference[`KafkaListeners` schema reference].

NOTE: Your configuration of {ProductPlatformName} must support Ingress NetworkPolicies in order to use network policies in {ProductName}.
