Kubernetes service not working as expected with kafka

Question

I'm trying to setup a zookeeper and kafka as separate Kubernetes deployments/pods in a shared namespace. I've bootstraped a local K8s 1.8 with Calico via kubeadm on my Ubuntu sandbox...

For the Zookeeper, I'm using the image zookeeper:3.4 from hub.docker.com and I created a Kubernetes deployment and service, where I expose ports: 2181 2888 3888. Service name is zookeeper and I assume I should be able to use it by this hostname from the pods in the namespace.

For the Kafka 1.0, I've created my own container image, that I can control with environment variables... I'm setting the zookeeper.connect to zookeeper:2181. I assume the Kubernetes DNS will resolve this and open the connection to the service.

Unfortunately I get:

[2018-01-03 15:48:26,292] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
[2018-01-03 15:48:32,293] INFO Terminate ZkClient event thread. (org.I0Itec.zkclient.ZkEventThread)
[2018-01-03 15:48:46,286] INFO Opening socket connection to server zookeeper.sandbox.svc.cluster.local/10.107.41.148:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2018-01-03 15:48:46,299] INFO Socket connection established to zookeeper.sandbox.svc.cluster.local/10.107.41.148:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2018-01-03 15:48:46,319] INFO Session establishment complete on server zookeeper.sandbox.svc.cluster.local/10.107.41.148:2181, sessionid = 0x10000603c560001, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)
[2018-01-03 15:48:46,331] INFO Session: 0x10000603c560001 closed (org.apache.zookeeper.ZooKeeper)
[2018-01-03 15:48:46,333] FATAL Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer)
org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to zookeeper server 'zookeeper:2181' with timeout of 6000 ms
    at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1233)
    at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:157)
    at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:131)
    at kafka.utils.ZkUtils$.createZkClientAndConnection(ZkUtils.scala:115)
    at kafka.utils.ZkUtils$.withMetrics(ZkUtils.scala:92)
    at kafka.server.KafkaServer.initZk(KafkaServer.scala:346)
    at kafka.server.KafkaServer.startup(KafkaServer.scala:194)
    at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:38)
    at kafka.Kafka$.main(Kafka.scala:92)
    at kafka.Kafka.main(Kafka.scala)

So I was assuming I have a generic networking issue in my cluster, then I noticed something even more confusing for me... If I set zookeeper.connect to 10.107.41.148:2181 ( the current address of the zookeeper service ), the connection works ( at least from kafka to zookeeper ).

[2018-01-03 15:51:31,092] INFO Waiting for keeper state SyncConnected (org.I0Itec.zkclient.ZkClient)
[2018-01-03 15:51:31,094] INFO Opening socket connection to server 10.107.41.148/10.107.41.148:2181. Will not attempt to authenticate using SASL (unknown error) (org.apache.zookeeper.ClientCnxn)
[2018-01-03 15:51:31,105] INFO Socket connection established to 10.107.41.148/10.107.41.148:2181, initiating session (org.apache.zookeeper.ClientCnxn)
[2018-01-03 15:51:31,134] INFO Session establishment complete on server 10.107.41.148/10.107.41.148:2181, sessionid = 0x10000603c560005, negotiated timeout = 6000 (org.apache.zookeeper.ClientCnxn)

With this setup I'm able to use the zookeeper service from the host of the kubernetes cluster to do for example "bin/kafka-topics.sh --list --zookeeper 10.107.41.148:2181". Producing a message does not work thou... I assume once the network is properly working, I need to add the kafka advertised adddress ...

kafka-console-producer.sh --broker-list 10.100.117.196:9092 --topic test1
>test-msg1
>[2018-01-03 17:05:35,689] WARN [Producer clientId=console-producer] Connection to node 0 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

Any hints what is wrong with my Kubernetes network setup or at least where to start troubleshooting?

Thank you and best regards, Pavel

Answer 1

If you use statefulsets, you need to deploy the service first.

Here my service.

apiVersion: v1
kind: Service
metadata:
  name: zookeeper
  labels:
    app: zookeeper
spec:
  clusterIP: None
  ports:
  - port: 2181
    name: client
  - port: 2888
    name: server
  - port: 3888
    name: leader-election
  selector:
    app: zookeeper

Here the configmap (used later):

 apiVersion: v1
    kind: ConfigMap
    metadata:
      name: zookeeper-cm
    data:
      jvm.heap: "1G"
      tick: "2000"
      init: "10"
      sync: "5"
      client.cnxns: "60"
      snap.retain: "3"
      purge.interval: "0"

Here my statefulset:

apiVersion: apps/v1beta1
kind: StatefulSet
metadata:
  name: zookeeper
spec:
  serviceName: zookeeper
  replicas: 1
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: "app"
                    operator: In
                    values:
                    - zookeeper
              topologyKey: "kubernetes.io/hostname"
      containers:
      - name: zookeeper
        imagePullPolicy: IfNotPresent
        image: sorintdev/zookeeper:20171204m
        resources:
          requests:
            memory: 500Mi
            cpu: 200m
        ports:
        - containerPort: 2181
          name: client
        - containerPort: 2888
          name: server
        - containerPort: 3888
          name: leader-election
        env:
        - name : ZK_REPLICAS
          value: "1"
        - name : ZK_HEAP_SIZE
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: jvm.heap
        - name : ZK_TICK_TIME
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: tick
        - name : ZK_INIT_LIMIT
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: init
        - name : ZK_SYNC_LIMIT
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: tick
        - name : ZK_MAX_CLIENT_CNXNS
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: client.cnxns
        - name: ZK_SNAP_RETAIN_COUNT
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: snap.retain
        - name: ZK_PURGE_INTERVAL
          valueFrom:
            configMapKeyRef:
                name: zookeeper-cm
                key: purge.interval
        - name: ZK_CLIENT_PORT
          value: "2181"
        - name: ZK_SERVER_PORT
          value: "2888"
        - name: ZK_ELECTION_PORT
          value: "3888"
        command:
        - bash
        - -c
        - zkGenConfig.sh && zkServer.sh start-foreground
        readinessProbe:
          exec:
            command:
            - "zkOk.sh"
          initialDelaySeconds: 10
          timeoutSeconds: 5
        livenessProbe:
          exec:
            command:
            - "zkOk.sh"
          initialDelaySeconds: 10
          timeoutSeconds: 5
        volumeMounts:
        - name: data
          mountPath: /var/lib/zookeeper
      securityContext:
        runAsUser: 1000
        fsGroup: 1000
  volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      resources:
        requests:
          storage: 1Gi
      accessModes:
      - ReadWriteOnce
      storageClassName: zookeeper-class

After you deployed a working zookeeper configuration and they have elected a master, you can proceed with kafka deployment.

Once zookeeper is deployed, your kafka configuration must reference to zookeeper statefulset through the service. In kafka you must define/override this property:

--override zookeeper.connect=zookeeper-0.zookeeper:2181

From inside a pod you should ping zookeeper-0.zookeeper succesfully.

Hope this helps.