<p>I have a Node.js microservice connecting to a Kafka service. If the Kafka broker is in the default namespace with the other microservices, there are no issues.</p>
<pre><code>const kafka = new Kafka({
  clientId: 'my-app',
  // brokers: ['kafka-broker.kafka.svc.cluster.local:9092'],
  brokers: ['kafka-broker-srv:9092'],
})

const admin = kafka.admin()
    await admin.connect()
    console.log('admin connected')
    console.log('TOPICS', await admin.listTopics())
</code></pre>
<pre><code>[auth-pod] admin connected
[auth-pod] TOPICS [ 'test-topic' ]
</code></pre>
<p>Here is the Kubernetes configuration:</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-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
          # automatically create topics
        - name: KAFKA_CREATE_TOPICS
          value: 'test-topic:1:1'
        - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
          value: 'true'
        - name: KAFKA_DELETE_TOPIC_ENABLE
          value: 'true'
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
        - containerPort: 9092
</code></pre>
<p>However, if it is in a separate namespace, I can connect to the service <code>'kafka-broker-srv.kafka.svc.cluster.local:9092'</code> that is in the kafka namespace, but I get a <code>KafkaJSBrokerNotFound</code> error.</p>
<pre><code>[auth-pod] admin connected
[auth-pod] KafkaJSNonRetriableError
[auth-pod]   Caused by: KafkaJSBrokerNotFound: No brokers in the broker pool
[auth-pod]     at BrokerPool.withBroker (/app/node_modules/kafkajs/src/cluster/brokerPool.js:270:13)
[auth-pod]     ... 3 lines matching cause stack trace ...
[auth-pod]     at async test (/app/src/kakfka/connect.js:18:27) {
[auth-pod]   name: 'KafkaJSNumberOfRetriesExceeded',
[auth-pod]   retriable: false,
[auth-pod]   helpUrl: undefined,
[auth-pod]   retryCount: 5,
[auth-pod]   retryTime: 12820,
[auth-pod]   [cause]: KafkaJSBrokerNotFound: No brokers in the broker pool
[auth-pod]       at BrokerPool.withBroker (/app/node_modules/kafkajs/src/cluster/brokerPool.js:270:13)
[auth-pod]       at /app/node_modules/kafkajs/src/cluster/index.js:191:32
[auth-pod]       at process.processTicksAndRejections (node:internal/process/task_queues:95:5)
[auth-pod]       at async Object.listTopics (/app/node_modules/kafkajs/src/admin/index.js:103:31)
[auth-pod]       at async test (/app/src/kakfka/connect.js:18:27) {
[auth-pod]     retriable: true,
[auth-pod]     helpUrl: undefined,
[auth-pod]     [cause]: undefined
[auth-pod]   }
[auth-pod] }
</code></pre>
<p>I don't see what the issue could be since it worked before and the only thing that would need to change is to specify the namespace to connect to the service <code>.kafka.svc.cluster.local</code>.</p>
<p>If I use the same name for the service and the broker <code>'kafka-broker.kafka.svc.cluster.local:9092'</code> (I thought I wasn't supposed to use difference names), I get a connection timeout.</p>
<pre><code>[auth-pod] admin connected
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:11:57.662Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:11:59.139Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:00.351Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:02.154Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:02.958Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:03.230Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:05.359Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:05.461Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:06.533Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:09.139Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:09.640Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:10.646Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:11.536Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:15.262Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:16.151Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:17.160Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection timeout&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;}
[auth-pod] KafkaJSNonRetriableError
[auth-pod]   Caused by: KafkaJSConnectionError: Connection timeout
[auth-pod]     at Timeout.onTimeout [as _onTimeout] (/app/node_modules/kafkajs/src/network/connection.js:223:23)
[auth-pod]     at listOnTimeout (node:internal/timers:569:17)
[auth-pod]     at process.processTimers (node:internal/timers:512:7) {
[auth-pod]   name: 'KafkaJSNumberOfRetriesExceeded',
[auth-pod]   retriable: false,
[auth-pod]   helpUrl: undefined,
[auth-pod]   retryCount: 5,
[auth-pod]   retryTime: 10402,
[auth-pod]   [cause]: KafkaJSConnectionError: Connection timeout
[auth-pod]       at Timeout.onTimeout [as _onTimeout] (/app/node_modules/kafkajs/src/network/connection.js:223:23)
[auth-pod]       at listOnTimeout (node:internal/timers:569:17)
[auth-pod]       at process.processTimers (node:internal/timers:512:7) {
[auth-pod]     retriable: true,
[auth-pod]     helpUrl: undefined,
[auth-pod]     broker: 'kafka-broker:9092',
[auth-pod]     code: undefined,
[auth-pod]     [cause]: undefined
[auth-pod]   }
[auth-pod] }
[auth-pod] {&quot;level&quot;:&quot;ERROR&quot;,&quot;timestamp&quot;:&quot;2023-03-28T04:12:17.548Z&quot;,&quot;logger&quot;:&quot;kafkajs&quot;,&quot;message&quot;:&quot;[Connection] Connection error: getaddrinfo EAI_AGAIN kafka-broker&quot;,&quot;broker&quot;:&quot;kafka-broker:9092&quot;,&quot;clientId&quot;:&quot;my-app&quot;,&quot;stack&quot;:&quot;Error: getaddrinfo EAI_AGAIN kafka-broker\n    at GetAddrInfoReqWrap.onlookup [as oncomplete] (node:dns:107:26)&quot;}
</code></pre>
<p>The same thing happens with producers and consumers, the <code>connect</code> call works, but not the subsequent methods.</p>
<pre><code>// producer
const producer = kafka.producer()
await producer.connect()
console.log('producer connected', producer)
send
await producer.send({
  topic: 'test-topic',
  messages: [{ value: 'Hello KafkaJS user!' }],
})
// await producer.disconnect()

// consumer
const consumer = kafka.consumer({ groupId: 'test-group' })
await consumer.connect()
console.log('consumer connected', consumer)
// receive
await consumer.subscribe({ topic: 'test-topic', fromBeginning: true })
await consumer.run({
  eachMessage: async ({ topic, partition, message }) =&gt; {
    console.log('CONSUME', {
      value: message.value.toString(),
    })
  },
})
</code></pre>
<p>I am not sure why the naming of the service would make a difference.</p>
<p>Given that I can connect to the service and that the dns resolution is the same within any namespace (so I doubt I would need to change any names in the kafka configuration), what could cause these errors? <a href="https://github.com/bogdan-pechounov/microservices-quiz-app/tree/f63eb96696683cdb7276203ea5c2e32280d1f1db" rel="nofollow noreferrer">full code</a></p>
<hr />
<p>Interestingly, if I deploy to both namespaces and connect to the service in the kafka namespace, it does work.</p>
<pre><code>  brokers: ['kafka-broker-srv.kafka.svc.cluster.local:9092'],
</code></pre>
<pre><code>[auth-pod] admin connected
[auth-pod] TOPICS [ 'test-topic' ]
</code></pre>
<p>Is kafkajs connecting to the broker in the default namespace directly; if so, what is the point of going through the kafka service (in the kafka namespace)?</p>
<p>Edit:
Since I am using a headless service, the phrasing &quot;going through the kafka service&quot; probably doesn't make sense. The service name is resolved to the ip of the pod directly. If <code>connect</code> works, does that mean the connection to the pod succeeded; if there is a connection error for subsequent method calls, is kafkajs connecting to something other than the pod?</p>

BPDev
<p>You've set this</p>
<pre><code>- name: KAFKA_ADVERTISED_LISTENERS
  value: PLAINTEXT://kafka-broker:9092
</code></pre>
<p>This means that <strong>after</strong> a client connects via the bootstrap protocol (when given the FQDN), it sees <strong>this</strong> address (and not your headless service), not the value you provided in the app code.<br />
This has no reference to any namespace, therefore will try to lookup <code>tcp://kafka-broker<b>.default.svc.cluster.local</b>:9092</code> (assuming your app is in the <code>default</code> namespace).</p>
<p>For more details, refer <a href="https://stackoverflow.com/questions/51630260/connect-to-kafka-running-in-docker">Connect to Kafka running in Docker</a></p>
<p>As <a href="https://stackoverflow.com/a/75844920/2308683">mentioned before</a>, you need to advertise the FQDN / address of that <em>specific</em> Kafka pod, so that clients can connect directly to exact brokers, and not to a Service address that is round-robin / load-balanced. For example, Strimzi or Bitnami Helm Charts will advertise values like <code>kafka-0</code> and <code>kafka-1</code> for a two-node cluster, while still maintaining some <code>kafka-svc</code> Headless Service to use for bootstrapping.</p>
<hr />
<blockquote>
<p>if I deploy to both namespaces and connect to the service in the kafka namespace, it does work</p>
</blockquote>
<p>Because that is <a href="https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/" rel="nofollow noreferrer">how DNS lookups are supposed to work</a>.</p>
