Kafka with Ingress TLS (NotLeaderOrFollowerException)

[tarunmittal24]

Hi i am trying to setup the simple configuration kafka cluster with ingress, below is my file:-

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
spec:
  entityOperator:
    topicOperator: {}
    userOperator: {}
  kafka:  
    authorization:
       type: simple
       superUsers:
         - CN=admin-user
    version: 3.5.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9094
        type: ingress
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
            host: kafka.beta.example.com
            alternativeNames:
              - kafka.beta.example.com
              - kafka-0.beta.example.com
              - kafka-1.beta.example.com
              - kafka-2.beta.example.com
              - kafka-cluster-kafka-tls-bootstrap.kafka.svc
          brokers:
          - broker: 0
            host: kafka-0.beta.example.com
            advertisedHost: kafka-0.beta.example.com
            advertisedPort: 9094
          - broker: 1
            host: kafka-1.beta.example.com
            advertisedHost: kafka-1.beta.example.com
            advertisedPort: 9094
          - broker: 2
            host: kafka-2.beta.example.com
            advertisedHost: kafka-2.beta.example.com
            advertisedPort: 9094
          class: internal-tcp    
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: "2.4"
    storage:
      type: persistent-claim
      size: 2Gi
      deleteClaim: true
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 1Gi
      deleteClaim: true

and also i create one user:-

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaUser
metadata:
  namespace: kafka
  name: app-user
  labels:
    strimzi.io/cluster: kafka-cluster
spec:
  authentication:
    type: tls
  authorization:
    type: simple
    acls:
      - resource:
          type: topic
          name: "*"
        operation: All
      - resource:
          type: group
          name: "*"
        operation: All
      - resource:
          type: cluster
          name: "kafka-cluster"
          patternType: literal
        operation: All
      - resource:
          type: transactionalId
          name: "*"
          patternType: literal
        operation: All

Services Created by the above config

Now from inside the pod, when i am trying to access the kafka, using the kafka-cluster-kafka-tls-bootstrap service or with the kafka.beta.example.com

client-ssl.properties

security.protocol=SSL
ssl.keystore.location=/user-keystore.jks
ssl.keystore.password=iN0l0dcex89o0fTy**************
ssl.key.password=iN0l0dcex89o0fTyD************
ssl.truststore.location=/user-truststore.jks
ssl.truststore.password=iN0l0dcex89o0fT***************

kafka-console-producer.sh --broker-list kafka.beta.example.com:9094 --topic test-topic --producer.config client-ssl.properties

or

kafka-console-producer.sh --broker-list kafka-cluster-kafka-tls-bootstrap:9094 --topic test-topic --producer.config client-ssl.properties

I am getting the following error

23[2023-12-29 16:39:15,391] WARN [Producer clientId=console-producer] Got error produce response with correlation id 6 on topic-partition test-topic-2, retrying (2 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:15,392] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
4[2023-12-29 16:39:15,533] WARN [Producer clientId=console-producer] Got error produce response with correlation id 8 on topic-partition test-topic-2, retrying (1 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:15,534] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

>[2023-12-29 16:39:15,914] WARN [Producer clientId=console-producer] Got error produce response with correlation id 10 on topic-partition test-topic-2, retrying (0 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:15,915] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,057] ERROR Error when sending message to topic test-topic with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.
[2023-12-29 16:39:16,058] ERROR Error when sending message to topic test-topic with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.
[2023-12-29 16:39:16,058] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,589] WARN [Producer clientId=console-producer] Got error produce response with correlation id 14 on topic-partition test-topic-2, retrying (2 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,590] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,734] WARN [Producer clientId=console-producer] Got error produce response with correlation id 16 on topic-partition test-topic-2, retrying (1 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,734] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,874] WARN [Producer clientId=console-producer] Got error produce response with correlation id 18 on topic-partition test-topic-2, retrying (0 attempts left). Error: NOT_LEADER_OR_FOLLOWER (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:16,874] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)
[2023-12-29 16:39:17,017] ERROR Error when sending message to topic test-topic with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.
[2023-12-29 16:39:17,018] WARN [Producer clientId=console-producer] Received invalid metadata error in produce request on partition test-topic-2 due to org.apache.kafka.common.errors.NotLeaderOrFollowerException: For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.. Going to request metadata update now (org.apache.kafka.clients.producer.internals.Sender)

[tarunmittal24]

@scholzj , can you please help here.

[tarunmittal24]

Thanks @scholzj , for the reply, it will perfectly work if we use the nginx ingress, but if we use the different ingress-controller like in my case i am using kong, it is an issue.

Issue in my setup is that, my load-balancer is not distributing the traffic correctly and causing the issue, and why it is also not working from inside the k8's cluster, when i am using the service endpoint, it is using the same advertised host / port configuration, which is taking them directly to the nlb and that is causing the issue, because my kafka-0.beta.example.com, kafka-1.beta.example.com and kafka-2.beta.example.com entries are on aws route53, which is pointing to nlb

How I Identified:-

Inside the pod, i manually edit the /etc/hosts and put the respective service ip to each host. below is my file:-, Ip's is changed from the above screenshot, as i am destroying and creating the cluster so many times.

172.20.132.77   kafka.beta.example.com
172.20.32.93    kafka-0.beta.example.com
172.20.93.6      kafka-1.beta.example.com
172.20.78.198   kafka-2.beta.example.com

After this it's start working, and i came to know that this is the ingress routing issue.

My final Working config:-

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
spec:
  entityOperator:
    topicOperator: {}
    userOperator: {}
  kafka:  
    authorization:
       type: simple
       superUsers:
         - CN=admin-user
    version: 3.5.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9094
        type: cluster-ip
        tls: true
        authentication:
          type: tls
        configuration:
          bootstrap:
            alternativeNames:
              - kafka.beta.example.com
              - kafka-0.beta.example.com
              - kafka-1.beta.example.com
              - kafka-2.beta.example.com
              - kafka-cluster-kafka-tls-bootstrap.kafka.svc
          brokers:  
          - broker: 0
            advertisedHost: kafka-0.beta.example.com
            advertisedPort: 9095
          - broker: 1
            advertisedHost: kafka-1.beta.example.com
            advertisedPort: 9096
          - broker: 2
            advertisedHost: kafka-2.beta.example.com
            advertisedPort: 9097
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: "2.4"
    storage:
      type: persistent-claim
      size: 2Gi
      deleteClaim: true
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 1Gi
      deleteClaim: true

and my kong ingress config is :-

apiVersion: configuration.konghq.com/v1beta1
kind: TCPIngress
metadata:
  annotations:
    external-dns.alpha.kubernetes.io/dns-type: private
    external-dns.alpha.kubernetes.io/hostname: kafka.beta.example.com
    kubernetes.io/ingress.class: internal-tcp
  name: kafka-tcp-ing
spec:
  rules:
  - backend:
      serviceName: kafka-cluster-kafka-tls-bootstrap
      servicePort: 9094
    port: 9094
---
apiVersion: configuration.konghq.com/v1beta1
kind: TCPIngress
metadata:
  annotations:
    external-dns.alpha.kubernetes.io/dns-type: private
    external-dns.alpha.kubernetes.io/hostname: kafka-0.beta.example.com
    kubernetes.io/ingress.class: internal-tcp
  name: kafka-tcp-ing-0
spec:
  rules:
  - backend:
      serviceName: kafka-cluster-kafka-tls-0
      servicePort: 9094
    port: 9095
---
apiVersion: configuration.konghq.com/v1beta1
kind: TCPIngress
metadata:
  annotations:
    external-dns.alpha.kubernetes.io/dns-type: private
    external-dns.alpha.kubernetes.io/hostname: kafka-1.beta.example.com
    kubernetes.io/ingress.class: internal-tcp
  name: kafka-tcp-ing-1
spec:
  rules:
  - backend:
      serviceName: kafka-cluster-kafka-tls-1
      servicePort: 9094
    port: 9096
---
apiVersion: configuration.konghq.com/v1beta1
kind: TCPIngress
metadata:
  annotations:
    external-dns.alpha.kubernetes.io/dns-type: private
    external-dns.alpha.kubernetes.io/hostname: kafka-2.beta.example.com
    kubernetes.io/ingress.class: internal-tcp
  name: kafka-tcp-ing-2
spec:
  rules:
  - backend:
      serviceName: kafka-cluster-kafka-tls-2
      servicePort: 9094
    port: 9097

I will write the blog on this to setup it on kong ingress, will post it here.

Thanks for the help @scholzj

[scholzj]

Yeah, sure with the way you modified the hosts file it will work from inside. But you are not using the Ingress anymore. So basically the way you would normally do it is to create an internal listener for use from inside Kubernetes and one for use from outside. 'm afraid I do not know Kong, so I cannot help much with that setup. But on the Strimzi side, you have it right now in the KAfka CR -> using the type: cluster-ip, overriding the advertised hostname / port and creating the Ingresses. That is the right way to go.

[tarunmittal24]

yes it is working perfectly.

[tarunmittal24]

Blog:- https://mittaltarun9715.medium.com/access-strimzi-kafka-using-ingress-94fd83acdebd

[scholzj]

That looks good. Please let me know if/once you share it on LinkedIn / Twitter etc., we can reshare it 😉.

One side note regarding the cost for the load balancers you mention there. What you say there is correct. But I think what is worth considering there is also what the pricing schema for the load balancers is, how you use the external listener, and how much traffic you send through it. Quite often the costs of the data transfers are much bigger than the fixed costs per-loadbalancer. So the relative savings for using Ingress with one load balancer might be quite small at the end. (You do not have to mention it in the blog post of course - I just wanted to mention it here)

[mescaja]

Not sure it'll fix your issue but i got the exact same error message pushing down events to a Kafka topic using kafka-console-producer.sh

I got it fixed after setting the JAVA_HOME to a jdk 17. I had JDK 24 setup by default

export JAVA_HOME=/home/admin/jdk-17.0.12
kafka-console-producer.sh --producer.config ./client.properties --bootstrap-server kafka.poc:9092 --topic test