如何通过nodeport访问Kafka

Question

我想使用 nodeport 访问我的 Kafka 集群。这是我的 CRD，我正在尝试使用 nodeport 公开 Kafka。

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.6.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: nodeport
        tls: false
    overrides:
      bootstrap:
        nodePort: 32100
      brokers:
      - broker: 0
        nodePort: 32000
      - broker: 1
        nodePort: 32001
      - broker: 2
        nodePort: 32002
    config:
      offsets.topic.replication.factor: 3
      transaction.state.log.replication.factor: 3
      transaction.state.log.min.isr: 2
      log.message.format.version: "2.6"
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
  zookeeper:
    replicas: 3
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}



ist@ist-1207:~$ kubectl get node ist-1207 -o=jsonpath='{range .status.addresses[*]}{.type}{"\t"}{.address}{"\n"}'

InternalIP  192.168.105.62

Hostname    ist-1207



ist@ist-1207:~$ kubectl exec my-cluster-kafka-0 -c kafka -it -n strimzi -- cat /tmp/strimzi.properties | grep advertised

advertised.listeners=REPLICATION-9091://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi.svc:9091,PLAIN-9092://my-cluster-kafka-0.my-cluster-kafka-brokers.strimzi.svc:9092,TLS-9093://192.168.105.62:31255

我匹配了运行 Kafka pod 的地址和 Kafka 代理公布的地址 ..两者都相同，但我仍然无法访问。 以下是服务：

   NAMESPACE     NAME                             TYPE        CLUSTER-IP       EXTERNAL-IP   PORT(S)                      AGE
default       kubernetes                       ClusterIP   10.96.0.1        <none>        443/TCP                      2d17h
kube-system   kube-dns                         ClusterIP   10.96.0.10       <none>        53/UDP,53/TCP,9153/TCP       2d17h
strimzi       my-cluster-kafka-bootstrap       ClusterIP   10.97.105.228    <none>        9091/TCP,9092/TCP            2d15h
strimzi       my-cluster-kafka-brokers         ClusterIP   None             <none>        9091/TCP,9092/TCP            2d15h
strimzi       my-cluster-kafka-tls-0           NodePort    10.100.213.101   <none>        9093:31255/TCP               2d15h
strimzi       my-cluster-kafka-tls-1           NodePort    10.99.126.141    <none>        9093:30493/TCP               2d15h
strimzi       my-cluster-kafka-tls-2           NodePort    10.108.221.176   <none>        9093:30437/TCP               2d15h
strimzi       my-cluster-kafka-tls-bootstrap   NodePort    10.100.212.113   <none>        9093:31091/TCP               2d15h
strimzi       my-cluster-zookeeper-client      ClusterIP   10.109.94.99     <none>        2181/TCP                     2d15h
strimzi       my-cluster-zookeeper-nodes       ClusterIP   None             <none>        2181/TCP,2888/TCP,3888/TCP   2d15h
strimzi       my-connect-cluster-connect-api   ClusterIP   10.101.91.208    <none>        8083/TCP                     2d16h

[kafka@my-cluster-kafka-0 kafka]$ bin/kafka-topics.sh --bootstrap-server 192.168.105.62:31255 --list

Error while executing topic command : org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1606116174727, tries=1, nextAllowedTryMs=1606116174828) timed out at 1606116174728 after 1 attempt(s)

[2020-11-23 07:22:54,743] ERROR java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1606116174727, tries=1, nextAllowedTryMs=1606116174828) timed out at 1606116174728 after 1 attempt(s)

    at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)

    at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)

    at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)

    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)

    at kafka.admin.TopicCommand$AdminClientTopicService.getTopics(TopicCommand.scala:352)

    at kafka.admin.TopicCommand$AdminClientTopicService.listTopics(TopicCommand.scala:260)

    at kafka.admin.TopicCommand$.main(TopicCommand.scala:66)

    at kafka.admin.TopicCommand.main(TopicCommand.scala)

Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listTopics, deadlineMs=1606116174727, tries=1, nextAllowedTryMs=1606116174828) timed out at 1606116174728 after 1 attempt(s)

Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

 (kafka.admin.TopicCommand$)  

我卡在这里。我无法访问。如果我做错了什么，请帮助我。

Answer 1

我认为问题在于您在 nodeport 侦听器上配置了启用的 TLS，但不认为您已经提取了集群 CA 证书并按照官方文档中的描述在 kafka-topics 客户端上配置了信任库。 如果您不需要 TLS，只需在节点端口侦听器上使用 tls: false 禁用它。 您还可以在此博客文章中阅读有关使用 nodeport 的更多信息：

https://strimzi.io/blog/2019/04/23/accessing-kafka-part-2/

不要担心它使用旧的方式来定义监听器；你现在的那个是对的。