Kafka proxy getting "Failed to initialize a channel" error.

[rashiga]

I am trying to run a kafka cluster(v3.0.0) with 3 kafka replicas and 1 zookeeper. On top it we are using the Kafka proxy to setup ingress to allow remote access to kafka cluster. The config is as follows:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafkaproxy
spec:
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: kafkaproxy
  strategy:
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
    type: RollingUpdate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: kafkaproxy
    spec:
      containers:
      - env:
        - name: KAFKAPROXY_HOSTNAME
          value: kafkaproxy.test.com
        - name: KAFKAPROXY_BASE_PORT
          value: "xxxx"
        - name: KAFKAPROXY_BOOTSTRAP_SERVERS
          value: kafka-kafka-bootstrap.kafka-test.svc:9092
        image: dajudge/kafkaproxy:0.0.18
        imagePullPolicy: IfNotPresent
        name: sampleproxy
        ports:
        - containerPort: 4000
          protocol: TCP
        resources:
           limits:
             cpu: 500m
             memory: 2Gi
           requests:
             cpu: 500m
             memory: 1Gi
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext:
        runAsUser: 0
      terminationGracePeriodSeconds: 30

I have also created a service for the proxy as follows:

apiVersion: v1
kind: Service
metadata:
  name: kafkaproxy
spec:
  ports:
  - name: bootstrap
    port: xxxx
    protocol: TCP
    targetPort: xxxx
  - name: broker-0
    port: xxxx+1
    protocol: TCP
    targetPort: xxxx+1
  - name: broker-1
    port: xxxx+2
    protocol: TCP
    targetPort: xxxx+2
  - name: broker-2
    port: xxxx+3
    protocol: TCP
    targetPort: xxxx+3
  selector:
    app: kafkaproxy
  sessionAffinity: None
  type: ClusterIP

When we are trying to send messages to Kafka cluster using the kafka-proxy hostname and port (kafkaproxy.test.com:xxxx) , We are getting errors as it is not able to connect to Kafka bootstrap service and is unable to get brokers to connect.

The error logs in Kafka proxy are:

        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

14:06:27 WARN  [io.ne.ch.ChannelInitializer] (nioEventLoopGroup-3-3) [] Failed to initialize a channel. Closing: [id: 0x7fc14273, L:/10.233.97.180:5060 - R:/100.65.159.39:52216]: io.netty.channel.ConnectTimeoutException: connection timed out: kafka-kafka-1.kafka-kafka-brokers.kafka1-test.svc/10.233.82.82:9092
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

14:06:27 WARN  [io.ne.ch.ChannelInitializer] (nioEventLoopGroup-3-2) [] Failed to initialize a channel. Closing: [id: 0xbfcc2bd1, L:/10.233.97.180:5058 - R:/10.233.76.0:38834]: io.netty.channel.ConnectTimeoutException: connection timed out: kafka-kafka-0.kafka-kafka-brokers.kafka-test.svc/10.233.82.81:9092
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

14:06:43 WARN  [io.ne.ch.ChannelInitializer] (nioEventLoopGroup-3-4) [] Failed to initialize a channel. Closing: [id: 59.39:42290]: io.netty.channel.ConnectTimeoutException: connection timed out: kafka-kafka-2.kafka-kafka-brokers.kafka1
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

14:06:57 WARN  [io.ne.ch.ChannelInitializer] (nioEventLoopGroup-3-3) [] Failed to initialize a channel. Closing: [id: 5.0:32816]: io.netty.channel.ConnectTimeoutException: connection timed out: kafka-kafka-2.kafka-kafka-brokers.kafka1-n
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

14:06:57 WARN  [io.ne.ch.ChannelInitializer] (nioEventLoopGroup-3-2) [] Failed to initialize a channel. Closing: [id: 5.0:45758]: io.netty.channel.ConnectTimeoutException: connection timed out: kafka-kafka-1.kafka-kafka-brokers.kafka1-n
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe$1.run(AbstractNioChannel.java:261)
        at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
        at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
        at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:164)
        at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:829)
        at com.oracle.svm.core.thread.JavaThreads.threadStartRoutine(JavaThreads.java:567)
        at com.oracle.svm.core.posix.thread.PosixJavaThreads.pthreadStartRoutine(PosixJavaThreads.java:192)

It is getting disconnected with Kafka bootstrap even if we are flushing in messages continuously thrugh the producer.