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

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

.Prerequisites

* An {ProductPlatformName} cluster
* A running Cluster Operator

.Procedure

. Deploy Kafka cluster with an external listener enabled and configured to the type `nodeport`.
+
An example configuration with an external listener configured to use node ports:
+
[source,yaml,subs=attributes+]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
spec:
  kafka:
    # ...
    listeners:
      external:
        type: nodeport
        # ...
    # ...
  zookeeper:
    # ...
----

. Create or update the resource.
+
ifdef::Kubernetes[]
On {KubernetesName} this can be done using `kubectl apply`:
[source,shell,subs=+quotes]
kubectl apply -f _<your-file>_
+
endif::Kubernetes[]
On {OpenShiftName} this can be done using `oc apply`:
+
[source,shell,subs=+quotes]
oc apply -f _<your-file>_

. Find the port number of the bootstrap service.
+
ifdef::Kubernetes[]
On {KubernetesName} this can be done using `kubectl get`:
[source,shell,subs=+quotes]
kubectl get service _<cluster-name>_-kafka-external-bootstrap -o=jsonpath='{.spec.ports[0].nodePort}{"\n"}'
+
endif::Kubernetes[]
On {OpenShiftName} this can be done using `oc get`:
+
[source,shell,subs=+quotes]
oc get service _<cluster-name>_-kafka-external-bootstrap -o=jsonpath='{.spec.ports[0].nodePort}{"\n"}'
+
The port should be used in the Kafka _bootstrap_ address.

. Find the address of the {ProductPlatformName} node.
+
ifdef::Kubernetes[]
On {KubernetesName} this can be done using `kubectl get`:
[source,shell,subs=+quotes]
kubectl get node _<node-name>_ -o=jsonpath='{range .status.addresses[*]}{.type}{"\t"}{.address}{"\n"}'
+
endif::Kubernetes[]
On {OpenShiftName} this can be done using `oc get`:
+
[source,shell,subs=+quotes]
oc get node _<node-name>_ -o=jsonpath='{range .status.addresses[*]}{.type}{"\t"}{.address}{"\n"}'
+
If several different addresses are returned, select the address type you want based on the following order:
+
.. ExternalDNS
.. ExternalIP
.. Hostname
.. InternalDNS
.. InternalIP
+
Use the address with the port found in the previous step in the Kafka _bootstrap_ address.

. Extract the public certificate of the broker certification authority
+
ifdef::Kubernetes[]
On {KubernetesName} this can be done using `kubectl get`:
[source,shell,subs=+quotes]
kubectl get secret _<cluster-name>_-cluster-ca-cert -o jsonpath='{.data.ca\.crt}' | base64 -d > ca.crt
+
endif::Kubernetes[]
On {OpenShiftName} this can be done using `oc extract`:
+
[source,shell,subs=+quotes]
oc extract secret/_<cluster-name>_-cluster-ca-cert --keys=ca.crt --to=- > ca.crt
+
Use the extracted certificate in your Kafka client to configure TLS connection.
If you enabled any authentication, you will also need to configure SASL or TLS authentication.

.Additional resources
* For more information about the schema, see xref:type-KafkaListeners-reference[`KafkaListeners` schema reference].