K
Q

Can't access kafka from outside kubernetes

9/25/2020

I'm trying to acces kafka from outside kubernetes on my local machine. I'm using spring application to produce events on a topic. This is my deployment file for kafka:

kind: Deployment
metadata:
  name: kafka-broker0
  labels:
    app: kafka
spec:
  replicas: 2
  selector:
    matchLabels:
      app: kafka
      id: "0"
  template:
    metadata:
      labels:
        app: kafka
        id: "0"
    spec:
      containers:
      - name: kafka
        image: wurstmeister/kafka
        ports:
        - containerPort: 9092
        env:
        - name: KAFKA_ADVERTISED_PORT
          value: "30718"
        - name: KAFKA_ADVERTISED_HOST_NAME
          value: 192.168.1.240
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zoo1:2181
        - name: KAFKA_BROKER_ID
          value: "0"
        - name: KAFKA_CREATE_TOPICS
          value: LaunchScraper:1:1

And service file id:

kind: Service
metadata:
  name: kafka-services
  labels:
    name: kafka
spec:
  selector:
    app: kafka
    id: "0"
  ports:
    - protocol: TCP
      name: kafka-port
      port: 9092
  type: NodePort

I've allready created a zookeeper pod on kubernetes. My spring boot application shows this error:

2020-09-25 23:56:29.123  WARN 44324 --- [ad | producer-1] org.apache.kafka.clients.NetworkClient   : [Producer clientId=producer-1] Connection to node -1 (/192.168.1.240:9092) could not be established. Broker may not be available.
-- amine ben abdallah
apache-kafka
kubernetes

Similar Questions

Unable to create service account. "Already exists", but can't be found
Postgres vs Cassandra for Kong
Reverse Proxy Behavior in a K8s Cluster for SSH Calls
Facing an issue in setup docker distribution library using glide
Kubernetes's Ingress annotations for x509 certificate authentificate

2 Answers

5/3/2021

Deploying Kafka on Kubernetes was actually not as trivial as I first thought, but it worked after many trials and errors. Many examples you find on the internet did not work for me with the current version of Kubernetes / Kafka. What worked was:

  1. Using a StatefulSet for Kafka, not a Deployment
  2. setting KAFKA_ADVERTISED_LISTENERS and KAFKA_LISTENERS as shown below
  3. An additional port for outside access (in my case 32092, but is arbitrary) in the NodePort service, don't forget to access Kafka from outside via 32092 then, not 9092

A working example config would be (as replacement for your Deployment and Service, probably not minimal):

apiVersion: v1
kind: Service
metadata:
  labels:
    service: kafka
  name: kafka
spec:
  type: NodePort
  ports:
  - name: "9092"
    port: 9092
    protocol: TCP
    targetPort: 9092
  - name: "9093"
    port: 9093
    protocol: TCP
    targetPort: 9093
  - name: "32092"
    port: 32092
    protocol: TCP
    targetPort: 32092
    nodePort: 32092
  selector:
    service: kafka-instance
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    service: kafka-instance
  name: kafka-instance
spec:
  selector:
    matchLabels:
      service: kafka-instance
  serviceName: "kafka"
  replicas: 1
  template:
    metadata:
      labels:
        service: kafka-instance
    spec:
      containers:
      - env:
        - name: MY_HOST_IP
          valueFrom:
            fieldRef:
              fieldPath: status.hostIP
        - name: MY_POD_NAME
          valueFrom:
            fieldRef:
              fieldPath: metadata.name
        - name: KAFKA_ADVERTISED_LISTENERS
          value: INTERNAL://$(MY_POD_NAME).kafka.default.svc.cluster.local:9093,CLIENT://$(MY_POD_NAME).kafka.default.svc.cluster.local:9092,EXTERNAL://$(MY_HOST_IP):32092
        - name: KAFKA_INTER_BROKER_LISTENER_NAME
          value: INTERNAL
        - name: KAFKA_LISTENERS
          value: INTERNAL://:9093,CLIENT://:9092,EXTERNAL://:32092
        - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
          value: INTERNAL:PLAINTEXT,CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
        - name: KAFKA_PORT
          value: "9092"
        - name: KAFKA_RESTART_ATTEMPTS
          value: "10"
        - name: KAFKA_RESTART_DELAY
          value: "5"
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: zoo1:2181
        - name: KAFKA_ZOOKEEPER_SESSION_TIMEOUT
          value: "6000"
        - name: ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL
          value: "0"
        image: wurstmeister/kafka
        name: kafka-instance
        ports:
        - containerPort: 9092

If you don't already have a zookeeper, just add that and it should work:

apiVersion: v1
kind: Service
metadata:
  labels:
    service: zoo1
  name: zoo1
spec:
  ports:
  - name: "2181"
    port: 2181
    targetPort: 2181
  selector:
    service: zoo1-instance
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    service: zoo1-instance
  name: zoo1-instance
spec:
  selector:
    matchLabels:
      service: zoo1-instance
  serviceName: "zoo1"
  replicas: 1
  template:
    metadata:
      labels:
        service: zoo1-instance
    spec:
      containers:
      - image: wurstmeister/zookeeper
        name: zoo1-instance
        ports:
        - containerPort: 2181
-- dubi
Source: StackOverflow
9/26/2020

It seems like you've not fixed a nodePort in your service. When you make it to the value you've entered in KAFKA_ADVERTISED_PORT. Also set the KAFKA_ADVERTISED_HOST to your K8s node hostname/DNS.

In the spec for your sevice add nodePort: 30718 under the ports entry. Then in your client, try to connect on 30718 port using the node's address or hostname

Also, if you're looking to deploy Kafka on production, I'd recommend using operators like Strimzi https://Strimzi.io

-- tuxiedev
Source: StackOverflow