I want to access strimzi kafka cluster from outside kubernetes cluster.

[soominJung0413]

I'm having trouble seeing the Strimzi quickstart and applying it to my Kubernetes cluster.

The template is:

1. PV

apiVersion: v1
kind: PersistentVolume
metadata:
  labels:
    app: kafka-0
  name: kafka-0
spec:
  accessModes:
  - ReadWriteOnce
  capacity:
    storage: 20Gi
  storageClassName: local-storage
  persistentVolumeReclaimPolicy: Retain
  local:
    path: "/kafka-cluster/kafka/0/data"
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: "id"
          operator: In
          values:
          - "worker-2"
---
apiVersion: v1
kind: PersistentVolume
metadata:
  labels:
    app: zookeeper-0
  name: zookeeper-0
spec:
  accessModes:
  - ReadWriteOnce
  capacity:
    storage: 20Gi
  claimRef:
    apiVersion: v1
    kind: PersistentVolumeClaim
    name: data-kafka-cluster-zookeeper-0
    namespace: kafka-cluster
  storageClassName: local-storage
  persistentVolumeReclaimPolicy: Retain
  local:
    path: "/kafka-cluster/zookeeper/data"
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: "id"
          operator: In
          values:
          - "worker-2"

2. Kafka

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: kafka-cluster
spec:
  kafka:
    replicas: 1
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: external 
        port: 9093
        type: nodeport
        tls: false
        configuration:
          preferredNodePortAddressType: InternalIP
          bootstrap: 
            nodePort: 30721
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 10Gi
        class: local-storage
        selector:
          app: kafka-0
        deleteClaim: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 10Gi
      class: local-storage
      selector:
        app: zookeeper-0
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}

The deployment results are normal and the mount is fine.

After that, I created a topic.

apiVersion: kafka.strimzi.io/v1beta1
kind: KafkaTopic
metadata:
  name: my-topic
  namespace: kafka-cluster
  labels:
    strimzi.io/cluster: "kafka-cluster"
spec:
  partitions: 3
  replicas: 1

The WorkerNode is located in a Private SubNet due to the server structure, so I planned to access it from the outside through tunneling.
The broker's address came out to be accessible from outside the server, and the AWS security group was also released according to the node port.

telnet 127.0.0.1 4000 # NodePort 30721
# Trying 127.0.0.1...
# Connected to localhost.
# Escape character is '^]'.

telnet 127.0.0.1 4001 # NodePort 31736
# Trying 127.0.0.1...
# Connected to localhost.
# Escape character is '^]'

I tried to check the connection via telnet from outside the server and publish a message to a topic with local kafka, but there was a problem.

bin/kafka-console-producer.sh --broker-list 127.0.0.1:4000 \
--topic my-topic
# SLF4J: Class path contains multiple SLF4J bindings.
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/core/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/tools/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/api/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/transforms/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/runtime/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/file/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/mirror/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/mirror-client/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/json/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/.../kafka-2.8.0-src/connect/basic-auth-extension/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
# SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>hello

ERROR Error when sending message to topic my-topic with key: null, value: 5 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.TimeoutException: Topic my-topic not present in metadata after 60000 ms.

I don't know what I did wrong, I went through the quickstart and looked through the documentation, gathered the necessary properties, and looked for various resources, but I couldn't solve it.

p.s
I found that when I run local kafka on Public SubNet EC2, it works well. The IP:port used is the same. I don't know what causes this to happen. = 0
Is there any good way to tunnel from the outside..?

bin/kafka-console-producer.sh --broker-list [ip]:[nodeport] --topic my-topic

### Log ###
# SLF4J: Class path contains multiple SLF4J bindings.
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/core/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/tools/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/api/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/transforms/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/runtime/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/file/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/mirror/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/mirror-client/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/json/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/basic-auth-extension/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
# SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
>hello world
>hello messageSystem

### Subscribe
bin/kafka-console-consumer.sh --bootstrap-server [ip]:[nodeport] --topic my-topic --from-beginning

### Log ###
# SLF4J: Class path contains multiple SLF4J bindings.
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/core/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/tools/build/dependant-libs-2.13.5/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/api/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/transforms/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/runtime/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/file/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/mirror/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/mirror-client/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/json/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: Found binding in [jar:file:/root/strimzi/test/kafka-2.8.0-src/connect/basic-auth-extension/build/dependant-libs/slf4j-log4j12-1.7.30.jar!/org/slf4j/impl/StaticLoggerBinder.class]
# SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
# SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
hello world
hello messageSystem

Please take your time and help me. =(

[thuanghai]

I have encounted this problem some mouthes age. The zoo-entrance(https://github.com/scholzj/zoo-entrance) may help you. And you can also see #1337.

[scholzj]

Why are you using ports 4000 and 4001 when the node ports are 30721 and 31736? You should be using the node ports.

[soominJung0413]

This is because the EKS cluster has a private subnet, so when accessing the node port from the outside, it tunnels through the public subnet.

[scholzj]

Well, that will not work out of the box with Kafka then because it would screw up the advertised host names / ports and you would need to set them manually: https://strimzi.io/docs/operators/latest/full/using.html#type-GenericKafkaListenerConfigurationBroker-reference ... otherwise the discovery protocola Kafka uses will not work for the clients connecting through these tunnels.

[soominJung0413]

My EKS cluster uses alb-ingress-controller, so this is provisioning an ingress resource. I got the hint as in your answer and mapped Kafka's advertise.listener address to the broker service on the ingress resource, but I got a java heap memory exception because of an ssl problem. I don't think it's a good idea to take the aws certificate out and use it to connect to the broker. Is there any good way..?

[scholzj]

I'm not sure Application Loadbalancers would work with Kafka. They are designed for HTTP which Kafka isn't. And last time I checked, they did not support anything like TLS passthrough etc. So you might need to use either classic ELBs or NLBs.