<p>I can't connect to my kafka cluster from the outside. There seems to be a problem with the listeners and advertised listeners.</p>

<p>Any suggestions?</p>

<p><strong>When I try to connect from the outside on port 30092, then I always get a reference back to kafka-svc:9092</strong></p>

<ul>
<li>Cluster name: dev-docker-x02</li>
<li>How I test: default kafka for windows: .\bin\windows\kafka-topics.bat --list --bootstrap-server dev-docker-x02:30092</li>
<li>Requirement: use confluentinc/cp-kafka:5.4.0-1-ubi8</li>
</ul>

<p>My setup:</p>

<p><a href="https://i.stack.imgur.com/37LWw.jpg" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/37LWw.jpg" alt="enter image description here"></a></p>

<p>My broker configuration (The problem seems to be in the (advertised) listeners.</p>

<pre><code>kind: Deployment
metadata:
  name: kafka-deploy
spec:
  replicas: 1
  selector:
    matchLabels:
        app: kafka-pod
  template:
    metadata:
      labels:
        app: kafka-pod
    spec:
      containers:
      - name: kafka-ctr         # Container name
        image: confluentinc/cp-kafka:5.4.0-1-ubi8
        ports:
        - containerPort: 9092   # Port exposed by the container
        env:
        - name: KAFKA_BROKER_ID
          value: "0"
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zookeeper-svc:2181
        - name: KAFKA_LISTENERS
          value: "LISTENER_INTERNAL://:9092,LISTENER_EXTERNAL://:30092"
        - name: KAFKA_ADVERTISED_LISTENERS
          value: "LISTENER_INTERNAL://kafka-svc:9092,LISTENER_EXTERNAL://dev-kube-x02:30092"
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: "LISTENER_INTERNAL:PLAINTEXT,LISTENER_EXTERNAL:PLAINTEXT"
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: "LISTENER_EXTERNAL"
        - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
          value: "false"
        - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
          value: "1"
        - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
          value: "100"
</code></pre>

<p>Kafka has a lot of components like Headless Services, <a href="https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/#using-statefulsets" rel="nofollow noreferrer">Statefulsets</a> and each one has a distinctive role.
For that reason I'd suggest too the usage of <a href="https://github.com/confluentinc/cp-helm-charts/blob/master/charts/cp-kafka/README.md#external-access" rel="nofollow noreferrer">Kafka Confluentic Helm Chart</a>.</p>

<p>This guide is based on the helm chart since you mentioned you'd use it in the comments but the concepts here can be extended to any application that uses headless services and need external access.</p>

<p>For what you provided, I believe you are facing some difficulties because you are referencing a headless service externally, which will not work since the headless service does not have an internal operational IP.</p>

<blockquote>
  <p>The <a href="https://kubernetes.io/docs/concepts/services-networking/service/#headless-services" rel="nofollow noreferrer">Headless Service</a> is created alongside the StatefulSet. The created service will <strong>not</strong> be given a <code>clusterIP</code>, but will instead simply include a list of <code>Endpoints</code>.
   These <code>Endpoints</code> are then used to generate instance-specific DNS records in the form of:
    <code>&lt;StatefulSet&gt;-&lt;Ordinal&gt;.&lt;Service&gt;.&lt;Namespace&gt;.svc.cluster.local</code></p>
</blockquote>

<p>It creates a DNS name for each pod, e.g:</p>

<pre><code>[ root@curl:/ ]$ nslookup my-confluent-cp-kafka-headless
Server:    10.0.0.10
Address 1: 10.0.0.10 kube-dns.kube-system.svc.cluster.local

Name:      my-confluent-cp-kafka-headless
Address 1: 10.8.0.23 my-confluent-cp-kafka-1.my-confluent-cp-kafka-headless.default.svc.cluster.local
Address 2: 10.8.1.21 my-confluent-cp-kafka-0.my-confluent-cp-kafka-headless.default.svc.cluster.local
Address 3: 10.8.3.7 my-confluent-cp-kafka-2.my-confluent-cp-kafka-headless.default.svc.cluster.local
</code></pre>

<ul>
<li><p>This is what makes this services connect to each other inside the cluster.</p></li>
<li><p>You can't, therefore, expose <code>cp-kafka:9092</code> which is the headless service, also only used internally, as I explained above.</p></li>
<li>In order to get outside access <strong>you have to set the parameters <code>nodeport.enabled</code> to <code>true</code></strong> as stated here: <a href="https://github.com/confluentinc/cp-helm-charts/blob/master/charts/cp-kafka/README.md#external-access" rel="nofollow noreferrer">External Access Parameters</a>.</li>
<li>It adds one service to each kafka-N pod during chart deployment.</li>
<li>Note that the service created has the selector <code>statefulset.kubernetes.io/pod-name: demo-cp-kafka-0</code> this is how the service identifies the pod it is intended to connect to.</li>
</ul>

<hr>

<p><strong>Reproduction:</strong></p>

<ul>
<li><code>git clone https://github.com/confluentinc/cp-helm-charts.git</code></li>
<li>edit the file <code>cp-helm-charts/cp-kafka/values.yaml</code> changing the <code>nodeport</code> from <code>false</code> to <code>true</code> and change the ports as you'd like:</li>
</ul>

<pre><code>nodeport:
  enabled: true
  servicePort: 19092
  firstListenerPort: 31090
</code></pre>

<ul>
<li>Deploy the chart:</li>
</ul>

<pre><code>$ helm install demo cp-helm-charts
$ kubectl get pods
NAME                                       READY   STATUS    RESTARTS   AGE
demo-cp-control-center-6d79ddd776-ktggw    1/1     Running   3          113s
demo-cp-kafka-0                            2/2     Running   1          113s
demo-cp-kafka-1                            2/2     Running   0          94s
demo-cp-kafka-2                            2/2     Running   0          84s
demo-cp-kafka-connect-79689c5c6c-947c4     2/2     Running   2          113s
demo-cp-kafka-rest-56dfdd8d94-79kpx        2/2     Running   1          113s
demo-cp-ksql-server-c498c9755-jc6bt        2/2     Running   2          113s
demo-cp-schema-registry-5f45c498c4-dh965   2/2     Running   3          113s
demo-cp-zookeeper-0                        2/2     Running   0          112s
demo-cp-zookeeper-1                        2/2     Running   0          93s
demo-cp-zookeeper-2                        2/2     Running   0          74s

$ kubectl get svc
NAME                         TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)             AGE
demo-cp-control-center       ClusterIP   10.0.13.134   &lt;none&gt;        9021/TCP            50m
demo-cp-kafka                ClusterIP   10.0.15.71    &lt;none&gt;        9092/TCP            50m
demo-cp-kafka-0-nodeport     NodePort    10.0.7.101    &lt;none&gt;        19092:31090/TCP     50m
demo-cp-kafka-1-nodeport     NodePort    10.0.4.234    &lt;none&gt;        19092:31091/TCP     50m
demo-cp-kafka-2-nodeport     NodePort    10.0.3.194    &lt;none&gt;        19092:31092/TCP     50m
demo-cp-kafka-connect        ClusterIP   10.0.3.217    &lt;none&gt;        8083/TCP            50m
demo-cp-kafka-headless       ClusterIP   None          &lt;none&gt;        9092/TCP            50m
demo-cp-kafka-rest           ClusterIP   10.0.14.27    &lt;none&gt;        8082/TCP            50m
demo-cp-ksql-server          ClusterIP   10.0.7.150    &lt;none&gt;        8088/TCP            50m
demo-cp-schema-registry      ClusterIP   10.0.7.84     &lt;none&gt;        8081/TCP            50m
demo-cp-zookeeper            ClusterIP   10.0.9.119    &lt;none&gt;        2181/TCP            50m
demo-cp-zookeeper-headless   ClusterIP   None          &lt;none&gt;        2888/TCP,3888/TCP   50m
</code></pre>

<ul>
<li>My Node is on IP <code>35.226.189.123</code> and I'll connect to the <code>demo-cp-kafka-0-nodeport</code> nodeport service which is on port <code>31090</code>, now let's try to connect from outside the cluster. For that I'll connect to another VM where I have a minikube, so I can use <code>kafka-client</code> pod to test:</li>
</ul>

<pre><code>user@minikube:~$ kubectl get pods
NAME           READY   STATUS    RESTARTS   AGE
kafka-client   1/1     Running   0          17h

user@minikube:~$ kubectl exec kafka-client -it -- bin/bash

root@kafka-client:/# kafka-console-consumer --bootstrap-server 35.226.189.123:31090 --topic demo-topic --from-beginning --timeout-ms 8000 --max-messages 1
Wed Apr 15 18:19:48 UTC 2020
Processed a total of 1 messages
root@kafka-client:/# 
</code></pre>

<p>As you can see, I was able to access the kafka from outside.</p>

<ul>
<li>Using this method, the helm chart will create 1 external service for each replica you define.</li>
<li>If you need external access to Zookeeper it's not automatically provisioned like the kafka agent, but I'll leave a service model for you:</li>
</ul>

<p><code>zookeeper-external-0.yaml</code></p>

<pre><code>apiVersion: v1
kind: Service
metadata:
  labels:
    app: cp-zookeeper
    pod: demo-cp-zookeeper-0
  name: demo-cp-zookeeper-0-nodeport
  namespace: default
spec:
  externalTrafficPolicy: Cluster
  ports:
  - name: external-broker
    nodePort: 31181
    port: 12181
    protocol: TCP
    targetPort: 31181
  selector:
    app: cp-zookeeper
    statefulset.kubernetes.io/pod-name: demo-cp-zookeeper-0
  sessionAffinity: None
  type: NodePort
</code></pre>

<ul>
<li>It will create a service for it:</li>
</ul>

<pre><code>NAME                           TYPE        CLUSTER-IP    EXTERNAL-IP   PORT(S)             AGE
demo-cp-zookeeper-0-nodeport   NodePort    10.0.5.67     &lt;none&gt;        12181:31181/TCP     2s
</code></pre>

<ul>
<li>Test it with your external IP:</li>
</ul>

<pre><code>pod/zookeeper-client created
user@minikube:~$ kubectl exec -it zookeeper-client -- /bin/bash
root@zookeeper-client:/# zookeeper-shell 35.226.189.123:31181
Connecting to 35.226.189.123:31181
Welcome to ZooKeeper!
JLine support is disabled
</code></pre>

<p>If you have any doubts, let me know in the comments!</p>
