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

[id='proc-accessing-kafka-using-nodeports-{context}']
= Accessing Kafka using node ports

[role="_abstract"]
Use node ports to access a Kafka cluster from an external client outside the Kubernetes cluster.

To connect to a broker, you specify a hostname and port number for the Kafka bootstrap address, as well as the certificate used for TLS encryption.

The procedure shows basic `nodeport` listener configuration.
You can use listener properties to enable TLS encryption (`tls`) and specify a client authentication mechanism (`authentication`).
Add additional configuration using `configuration` properties.
For example, you can use the following configuration properties with `nodeport` listeners:

`preferredNodePortAddressType`:: Specifies the first address type that's checked as the node address. 
`externalTrafficPolicy`:: Specifies whether the service routes external traffic to node-local or cluster-wide endpoints.
`nodePort`:: Overrides the assigned node port numbers for the bootstrap and broker services.

For more information on listener configuration, see the link:{BookURLConfiguring}#type-GenericKafkaListener-reference[`GenericKafkaListener` schema reference^].

.Prerequisites

* A running Cluster Operator

In this procedure, the Kafka cluster name is `my-cluster`.
The name of the listener is `external4`.

.Procedure

. Configure a `Kafka` resource with an external listener set to the `nodeport` type.
+
For example:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  labels:
    app: my-cluster
  name: my-cluster
  namespace: myproject
spec:
  kafka:
    # ...
    listeners:
      - name: external4
        port: 9094
        type: nodeport
        tls: true
        authentication:
          type: tls
        # ...
    # ...
----

. Create or update the resource.
+
[source,shell,subs=+quotes]
kubectl apply -f _<kafka_configuration_file>_
+
A cluster CA certificate to verify the identity of the kafka brokers is created in the secret `my-cluster-cluster-ca-cert`.
+
`NodePort` type services are created for each Kafka broker, as well as an external bootstrap service.
+
.Node port services created for the bootstrap and brokers
[source,shell]
----
NAME                                  TYPE      CLUSTER-IP      PORT(S)               
my-cluster-kafka-external4-0          NodePort  172.30.55.13    9094:31789/TCP
my-cluster-kafka-external4-1          NodePort  172.30.250.248  9094:30028/TCP
my-cluster-kafka-external4-2          NodePort  172.30.115.81   9094:32650/TCP
my-cluster-kafka-external4-bootstrap  NodePort  172.30.30.23    9094:32650/TCP
----
+
The bootstrap address used for client connection is propagated to the `status` of the `Kafka` resource.
+
.Example status for the bootstrap address
[source,yaml,subs="+attributes"]
----
status:
  clusterId: Y_RJQDGKRXmNF7fEcWldJQ
  conditions:
    - lastTransitionTime: '2023-01-31T14:59:37.113630Z'
      status: 'True'
      type: Ready
  kafkaVersion: {DefaultKafkaVersion}
  listeners:
    # ...
    - addresses:
        - host: ip-10-0-224-199.us-west-2.compute.internal
          port: 32650
      bootstrapServers: 'ip-10-0-224-199.us-west-2.compute.internal:32650'
      certificates:
        - |
          -----BEGIN CERTIFICATE-----
          
          -----END CERTIFICATE-----
      name: external4
  observedGeneration: 2
  operatorLastSuccessfulVersion: {ProductVersion}
 # ...
----

. Retrieve the bootstrap address you can use to access the Kafka cluster from the status of the `Kafka` resource.
+
[source,shell,subs=+quotes]
----
kubectl get kafka my-cluster -o=jsonpath='{.status.listeners[?(@.name=="external4")].bootstrapServers}{"\n"}'

ip-10-0-224-199.us-west-2.compute.internal:32650
----

. Extract the cluster CA certificate.
+
[source,shell]
----
kubectl get secret my-cluster-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
----

. Configure your client to connect to the brokers.

.. Specify the bootstrap host and port in your Kafka client as the bootstrap address to connect to the Kafka cluster. For example, `ip-10-0-224-199.us-west-2.compute.internal:32650`.

.. Add the extracted certificate to the truststore of your Kafka client to configure a TLS connection.
+
If you enabled a client authentication mechanism, you will also need to configure it in your client.

NOTE: If you are using your own listener certificates, check whether you need to add the CA certificate to the client's truststore configuration. 
If it is a public (external) CA, you usually won't need to add it.

