<p>I have a Nodejs microservice and a Kafka broker running in the same cluster.</p>
<p>The kafka broker and zookeeper are running without errors, but I am not sure how to connect to them.</p>
<p><strong>kafka.yaml</strong></p>
<pre><code># create namespace
apiVersion: v1
kind: Namespace
metadata:
  name: &quot;kafka&quot;
  labels:
    name: &quot;kafka&quot;
---
# create zookeeper service
apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
  namespace: kafka
spec:
  type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - containerPort: 2181
---
# deploy kafka broker
apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-broker
  name: kafka-service
  namespace: kafka
spec:
  ports:
  - port: 9092
  selector:
    app: kafka-broker
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      hostname: kafka-broker
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: &quot;1&quot;
        - name: KAFKA_ZOOKEEPER_CONNECT
          # value: 10.244.0.35:2181
          value: zookeeper-service:2181
        - name: KAFKA_LISTENERS
          value: PLAINTEXT://:9092
        # - name: KAFKA_ADVERTISED_HOST_NAME
        #   value: kafka-broker
        # - name: KAFKA_ADVERTISED_PORT
        #   value: &quot;9092&quot;
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://kafka-broker:9092
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
        - containerPort: 9092
</code></pre>
<p><a href="https://levelup.gitconnected.com/how-to-deploy-apache-kafka-with-kubernetes-9bd5caf7694f" rel="nofollow noreferrer">source</a></p>
<p>Connecting using <code>kafka-service:9092</code> or <code>kafka-broker:9092</code> doesn't work and leads to a timeout.</p>
<p><strong>kafka.js</strong></p>
<pre><code>const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['PLAINTEXT://kafka-broker:9092'], // !!! connection string
})

async function createProducer() {
  const producer = kafka.producer()

  await producer.connect()
  await producer.send({
    topic: 'test-topic',
    messages: [{ value: 'Hello KafkaJS user!' }],
  })

  await producer.disconnect()
}

createProducer()
</code></pre>
<pre><code>[auth-pod] {&quot;level&quot;:&quot;WARN&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:41.511Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;KafkaJS v2.0.0 switched default partitioner. To retain the same partitioning behavior as in previous versions, create the producer with the option \&quot;createPartitioner: Partitioners.LegacyPartitioner\&quot;. See the migration guide at https://kafka.js.org/docs/migration-guide-v2.0.0#producer-new-default-partitioner for details. Silence this warning by setting the environment variable \&quot;KAFKAJS_NO_PARTITIONER_WARNING=1\&quot;&quot;}
[auth-pod] Listening on port 3000...
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:41.586Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:0,&quot;retryTime&quot;:292}     
[auth-pod] Connected to: mongodb://auth-mongo-srv:27017/auth
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:41.881Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:1,&quot;retryTime&quot;:596}     
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:42.479Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:2,&quot;retryTime&quot;:1184}    
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:43.665Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:3,&quot;retryTime&quot;:2782}    
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:46.449Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:4,&quot;retryTime&quot;:5562}    
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-24T15:35:52.015Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).&quot;,&quot;retryCount&quot;:5,&quot;retryTime&quot;:12506}   
[auth-pod] node:internal/process/promises:288
[auth-pod]             triggerUncaughtException(err, true /* fromPromise */);
[auth-pod]             ^
[auth-pod]
[auth-pod] KafkaJSNonRetriableError
[auth-pod]   Caused by: KafkaJSConnectionError: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).
[auth-pod]     at /app/node_modules/kafkajs/src/network/connection.js:254:11
[auth-pod]     ... 8 lines matching cause stack trace ...
[auth-pod]     at async createProducer (/app/src/kakfka/connect.js:11:3) {
[auth-pod]   name: 'KafkaJSNumberOfRetriesExceeded',
[auth-pod]   retriable: false,
[auth-pod]   helpUrl: undefined,
[auth-pod]   retryCount: 5,
[auth-pod]   retryTime: 12506,
[auth-pod]   [cause]: KafkaJSConnectionError: Failed to connect: Port should be &gt;= 0 and &lt; 65536. Received type number (NaN).
[auth-pod]       at /app/node_modules/kafkajs/src/network/connection.js:254:11
[auth-pod]       at new Promise (&lt;anonymous&gt;)
[auth-pod]       at Connection.connect (/app/node_modules/kafkajs/src/network/connection.js:167:12)
[auth-pod]       at ConnectionPool.getConnection (/app/node_modules/kafkajs/src/network/connectionPool.js:56:24)
[auth-pod]       at Broker.connect (/app/node_modules/kafkajs/src/broker/index.js:86:52)
[auth-pod]       at async /app/node_modules/kafkajs/src/cluster/brokerPool.js:93:9
[auth-pod]       at async /app/node_modules/kafkajs/src/cluster/index.js:107:14
[auth-pod]       at async Cluster.connect (/app/node_modules/kafkajs/src/cluster/index.js:146:5)
[auth-pod]       at async Object.connect (/app/node_modules/kafkajs/src/producer/index.js:219:7)
[auth-pod]       at async createProducer (/app/src/kakfka/connect.js:11:3) {
[auth-pod]     retriable: true,
[auth-pod]     helpUrl: undefined,
[auth-pod]     broker: 'PLAINTEXT:NaN',
[auth-pod]     code: undefined,
[auth-pod]     [cause]: undefined
[auth-pod]   }
[auth-pod] }
[auth-pod]
[auth-pod] Node.js v18.15.0
</code></pre>
<p>If I use the IP of the pod <code>kafka-broker-5c7f7d4f77-nxlwm</code> directly <code>brokers: ['10.244.0.94:9092']</code>, I also get an error. Using the default namespace instead of a separate namespace didn't make a difference.</p>
<p>After switching to a StatefulSet based on <a href="https://stackoverflow.com/a/57261043/20898396">this</a> answer, I can connect using the IP of <code>kafka-broker-0</code> <code>'10.244.0.110:9092'</code>, but I get another error: <code>KafkaJSProtocolError: Replication-factor is invalid</code>. I don't know why the dns resolution would fail, but using the name <code>'kafka-broker-0:9092'</code>, leads to the same error as before <code>&quot;[BrokerPool] Failed to connect to seed broker, trying another broker from the list: Connection timeout&quot;</code>.</p>
<p>Based on</p>
<blockquote>
<p>If you have multiple REST Proxy pods running, Kubernetes will route
the traffic to one of them. <a href="https://www.confluent.io/wp-content/uploads/Recommendations-for-Deploying-Apache-Kafka-on-Kubernetes.pdf" rel="nofollow noreferrer">source</a></p>
</blockquote>
<p>I should be able to use the Kubernetes service <code>kafka-service</code> to load balance requests without hard coding an IP address. (There wasn't a <code>targetPort</code>, but it still doesn't work after adding <code>targetPort: 9092</code>, although I am not sure which protocol to use)</p>
<hr />
<p>I looked at the logs of the kafka-broker pod and noticed an exception.</p>
<pre><code>[2023-03-24 18:01:25,123] WARN [Controller id=1, targetBrokerId=1] Error connecting to node kafka-broker:9092 (id: 1 rack: null) (org.apache.kafka.clients.NetworkClient)
java.net.UnknownHostException: kafka-broker
 at java.base/java.net.InetAddress$CachedAddresses.get(Unknown Source)
 at java.base/java.net.InetAddress.getAllByName0(Unknown Source)
 at java.base/java.net.InetAddress.getAllByName(Unknown Source)
 at java.base/java.net.InetAddress.getAllByName(Unknown Source)
 at org.apache.kafka.clients.DefaultHostResolver.resolve(DefaultHostResolver.java:27)
 at org.apache.kafka.clients.ClientUtils.resolve(ClientUtils.java:111)
 at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.currentAddress(ClusterConnectionStates.java:513)
 at org.apache.kafka.clients.ClusterConnectionStates$NodeConnectionState.access$200(ClusterConnectionStates.java:467)
 at org.apache.kafka.clients.ClusterConnectionStates.currentAddress(ClusterConnectionStates.java:172)
 at org.apache.kafka.clients.NetworkClient.initiateConnect(NetworkClient.java:985)
 at org.apache.kafka.clients.NetworkClient.ready(NetworkClient.java:311)
 at org.apache.kafka.clients.NetworkClientUtils.awaitReady(NetworkClientUtils.java:65)
 at kafka.controller.RequestSendThread.brokerReady(ControllerChannelManager.scala:292)
 at kafka.controller.RequestSendThread.doWork(ControllerChannelManager.scala:246)
 at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)
</code></pre>
<p>I think that specifying <code>KAFKA_ADVERTISED_LISTENERS</code> should be sufficient (<a href="https://stackoverflow.com/a/51632885/20610346">answer</a>), so I am guessing there is a problem with dns resolution.</p>
<p>Using a headless service by adding <code>clusterIP: &quot;None&quot;</code> and changing the name to <code>kafka-broker</code> in case that <code>PLAINTEXT://kafka-broker:9092</code> uses the service and not the deployment  didn't help.</p>
<pre><code># create namespace
apiVersion: v1
kind: Namespace
metadata:
  name: &quot;kafka&quot;
  labels:
    name: &quot;kafka&quot;
---
# create zookeeper service
apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
  namespace: kafka
spec:
  type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - containerPort: 2181
---
# deploy kafka broker
apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  clusterIP: &quot;None&quot;
  # ports:
  # - protocol: TCP
  #   port: 9092
  #   targetPort: 9092
  selector:
    app: kafka-broker
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  # replicas: 1
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      hostname: kafka-broker
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: &quot;1&quot;
        - name: KAFKA_ZOOKEEPER_CONNECT
          # value: 10.244.0.35:2181
          value: zookeeper-service:2181
        - name: KAFKA_LISTENERS
          value: PLAINTEXT://:9092
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://kafka-broker:9092
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
        - containerPort: 9092
</code></pre>
<p><a href="https://github.com/bogdan-pechounov/microservices-quiz-app/tree/f9c547cc160c2317222bd8bda99e10bff56818f6" rel="nofollow noreferrer">full code</a></p>
<p>Edit:
Not sure why I had a <code>KafkaJSProtocolError: Replication-factor is invalid</code> error, but changing the service as follows prevents it. (It might be because I was using the same name for the service and deployment. I don't fully understand headless services, but I also added a port.)</p>
<pre><code># create namespace
apiVersion: v1
kind: Namespace
metadata:
  name: &quot;kafka&quot;
  labels:
    name: &quot;kafka&quot;
---
# create zookeeper service
apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
  namespace: kafka
spec:
  # type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      # nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - containerPort: 2181
---
# deploy kafka broker
apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-srv
  name: kafka-srv
  namespace: kafka
spec:
  # headless service
  clusterIP: &quot;None&quot;
  ports:
  - name: foo
    port: 9092
  selector:
    app: kafka-broker
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  # replicas: 1
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      hostname: kafka-broker
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: &quot;1&quot;
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zookeeper-service:2181
        - name: KAFKA_LISTENERS
          value: PLAINTEXT://:9092
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://kafka-broker:9092
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
        - containerPort: 9092
</code></pre>
<pre><code>const { Kafka } = require('kafkajs')

const kafka = new Kafka({
  clientId: 'my-app',
  brokers: ['10.244.0.64:9092'],
})

async function createProducer() {
  const producer = kafka.producer()

  try {
    await producer.connect()
    console.log('connected', producer)
    // await producer.send({
    //   topic: 'test-topic',
    //   messages: [{ value: 'Hello KafkaJS user!' }],
    // })

    // await producer.disconnect()
  } catch (err) {
    console.log(&quot;Couldn' connect to broker&quot;)
    console.error(err)
  }
}
</code></pre>
<pre><code>[auth-pod] connected {
[auth-pod]   connect: [AsyncFunction: connect],
[auth-pod]   disconnect: [AsyncFunction: disconnect],
[auth-pod]   isIdempotent: [Function: isIdempotent],
[auth-pod]   events: {
[auth-pod]     CONNECT: 'producer.connect',
[auth-pod]     DISCONNECT: 'producer.disconnect',
[auth-pod]     REQUEST: 'producer.network.request',
[auth-pod]     REQUEST_TIMEOUT: 'producer.network.request_timeout',
[auth-pod]     REQUEST_QUEUE_SIZE: 'producer.network.request_queue_size'
[auth-pod]   },
[auth-pod]   on: [Function: on],
[auth-pod]   send: [AsyncFunction: send],
[auth-pod]   sendBatch: [AsyncFunction: sendBatch],
[auth-pod]   transaction: [AsyncFunction: transaction],
[auth-pod]   logger: [Function: getLogger]
[auth-pod] }
</code></pre>
<p>Edit 2: When connecting successfully using the ip address, I also get <code>java.net.UnknownHostException: kafka-broker</code> in the kafka-broker-0 pod. (The error keeps repeating as well) I thought that the pod was being reached by kafkajs and then the pod threw an error, but the error happens regardless. Matching the service name with the advertised host name prevents it.</p>
<pre><code>---
# deploy kafka broker
apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  # namespace: kafka
spec:
  # headless service
  clusterIP: &quot;None&quot;
  ports:
  - name: foo
    port: 9092
  selector:
    app: kafka-broker
</code></pre>
<p>I can't connect to the pod <code>kafka-broker-0:9092</code> directly, but now using the service name <code>kafka-broker:9092</code> works.</p>

BPDev
<p>From outside k8s, your app cannot resolve cluster DNS names. You'll need to use an ClusterIP / NodePort address (but you've set that to None). This will also require you to advertise that. Refer <a href="https://strimzi.io/blog/2019/04/17/accessing-kafka-part-1/" rel="nofollow noreferrer">https://strimzi.io/blog/2019/04/17/accessing-kafka-part-1/</a></p>
<ol>
<li><p>Never use IPs for services. This includes Zookeeper connect property for Kafka. Ref. <a href="https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/" rel="nofollow noreferrer">https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/</a></p>
</li>
<li><p>I highly recommend not writing your own Kafka/Zookeeper specs. Use a Helm Chart or Operator such as <a href="https://strimzi.io" rel="nofollow noreferrer">https://strimzi.io</a> - mentioned in blog above</p>
</li>
</ol>
