kubernetes之检查我的 Kafka 和 Zookeeper 功能和连接

xiaohuochai 2025年05月04日编程语言 41 0

作为我系统的一部分，我需要在 Kubernetes 之上使用 Kafka 和 Zookeeper 集群。我正在使用 Statefulset 来部署它们，并使用 headless 服务让 Kafka 的 Broker 能够与 Zookeeper 服务器通信。

集群似乎正在运行 -(通过键入 kubectl get pods)

NAME                      READY   STATUS    RESTARTS   AGE 
kafka-statefulset-0       1/1     Running   14         5d10h 
kafka-statefulset-1       1/1     Running   14         5d10h 
kafka-statefulset-2       1/1     Running   16         5d10h 
kafka-statefulset-3       1/1     Running   14         5d10h 
kafka-statefulset-4       1/1     Running   14         5d10h 
zookeeper-statefulset-0   1/1     Running   5          5d10h 
zookeeper-statefulset-1   1/1     Running   5          5d10h 
zookeeper-statefulset-2   1/1     Running   5          5d10h

我的问题是我不太明白如何检查他们是否可以正常通信。

我已经尝试过的- 我试过了-

 kafkacat -L -b kafka-statefulset-0.kafka headless.default.svc.cluster.local:9093

得到了-

% ERROR: Failed to acquire metadata: Local: Broker transport failure

我试过了-

kafkacat -b 172.17.0.10:9093 -t second_topic -P

得到了-

% ERROR: Local: Host resolution failure: kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093/0: Failed to resolve 'kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093': Name or service not known 
% ERROR: Local: Host resolution failure: kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093/1: Failed to resolve 'kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093': Name or service not known 
% ERROR: Local: Host resolution failure: kafka-statefulset-3.kafka-headless.default.svc.cluster.local:9093/3: Failed to resolve 'kafka-statefulset-3.kafka-headless.default.svc.cluster.local:9093': Name or service not known 
% ERROR: Local: Host resolution failure: kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093/4: Failed to resolve 'kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093': Name or service not known 
% ERROR: Local: Host resolution failure: kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093/2: Failed to resolve 'kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093': Name or service not known

但是当我跑的时候 -

kafkacat -L -b 172.17.0.10:9093

我明白了-

Metadata for all topics (from broker -1: 172.17.0.10:9093/bootstrap): 
 5 brokers: 
  broker 2 at kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093 
  broker 4 at kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093 
  broker 1 at kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093 
  broker 3 at kafka-statefulset-3.kafka-headless.default.svc.cluster.local:9093 
  broker 0 at kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093 
 4 topics: 
  topic "second_topic" with 1 partitions: 
    partition 0, leader 4, replicas: 4, isrs: 4

据我所知，我没有正确配置服务以从集群外部连接它们，但我可以从集群内部连接到它们。尽管我可以在集群内与他们通信，但我不断收到无法理解的错误。

例如(在 pod 的容器中安装 Kafkacat 并尝试与其他 Kafka 的代理通信)-

 > kubectl exec -it  kafka-statefulset-0 bash  
 > apt-get install kafkacat 
 > kafkacat -b kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093 -t TOPIC -P 
 > kafkacat -L -b kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093

我得到了正确的元数据，但最后出现错误 -

etadata for all topics (from broker -1: kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093/bootstrap): 
 5 brokers: 
  broker 2 at kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093 
  broker 4 at kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093 
  broker 1 at kafka-statefulset-1.kafka-headless.default.svc.cluster.local:9093 
  broker 3 at kafka-statefulset-3.kafka-headless.default.svc.cluster.local:9093 
  broker 0 at kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093 
 5 topics: 
  topic "TOPIC" with 1 partitions: 
    partition 0, leader 2, replicas: 2, isrs: 2 
  topic "second_topic" with 1 partitions: 
    partition 0, leader 4, replicas: 4, isrs: 4 
  topic "first_topic" with 1 partitions: 
    partition 0, leader 2, replicas: 2, isrs: 2 
  topic "nir_topic" with 1 partitions: 
    partition 0, leader 0, replicas: 0, isrs: 0 
  topic "first" with 1 partitions: 
    partition 0, leader 3, replicas: 3, isrs: 3 
%3|1581685918.022|FAIL|rdkafka#producer-0| kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093/0: Failed to connect to broker at kafka-statefulset-0.kafka-headless.default.svc.cluster.local:: Interrupted system call 
%3|1581685918.022|ERROR|rdkafka#producer-0| kafka-statefulset-0.kafka-headless.default.svc.cluster.local:9093/0: Failed to connect to broker at kafka-statefulset-0.kafka-headless.default.svc.cluster.local:: Interrupted system call 
%3|1581685918.022|FAIL|rdkafka#producer-0| kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093/2: Failed to connect to broker at kafka-statefulset-2.kafka-headless.default.svc.cluster.local:: Interrupted system call 
%3|1581685918.022|ERROR|rdkafka#producer-0| kafka-statefulset-2.kafka-headless.default.svc.cluster.local:9093/2: Failed to connect to broker at kafka-statefulset-2.kafka-headless.default.svc.cluster.local:: Interrupted system call 
%3|1581685918.023|FAIL|rdkafka#producer-0| kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093/4: Failed to connect to broker at kafka-statefulset-4.kafka-headless.default.svc.cluster.local:: Interrupted system call 
%3|1581685918.023|ERROR|rdkafka#producer-0| kafka-statefulset-4.kafka-headless.default.svc.cluster.local:9093/4: Failed to connect to broker at kafka-statefulset-4.kafka-headless.default.svc.cluster.local:: Interrupted system call

问题是什么云？我如何检查它们之间的通信并检查它们两者的功能正确性 - Kafka 和 Zookeeper？

一些关于系统的更多信息(如果需要我可以复制所有的 YAML 配置，但是它很长)-

有 Kafka 的 Statefulset 和 Zookeeper 的 Statefulset。
Kafka 的 Broker 可以通过端口 2181 与 zookeeper 对话。我的配置是 -

-override zookeeper.connect=zookeeper-statefulset-0.zookeeper-headless.default.svc.cluster.local:2181 ...

我还构建了 headless 服务，用于通过端口 9093 与每个 Kafka 的 Broker 对话
动物园管理员端口:

一个。 2181 - 客户端端口

2888 - 服务器端口

3888 - 领导人选举端口

5.运行的服务-

NAME                 TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)             AGE 
kafka-headless       ClusterIP   None            <none>        9093/TCP            5d10h 
kubernetes           ClusterIP   10.96.0.1       <none>        443/TCP             5d11h 
zookeeper-cs         ClusterIP   10.106.99.170   <none>        2181/TCP            5d11h 
zookeeper-headless   ClusterIP   None            <none>        2888/TCP,3888/TCP   5d11h

更新 - 添加 Kafka 和 Zookeeper 配置文件 卡夫卡-

apiVersion: apps/v1 
kind: StatefulSet 
metadata: 
spec: 
  selector: 
    matchLabels: 
      app: kafka-app 
  serviceName: kafka-headless 
  replicas: 5 
  podManagementPolicy: Parallel 
  template: 
    metadata: 
      labels: 
        app: kafka-app 
        kafkaMinReplicas: need-supervision 
    spec: 
      affinity: 
        podAffinity: 
          preferredDuringSchedulingIgnoredDuringExecution: 
             - weight: 1 
               podAffinityTerm: 
                 labelSelector: 
                    matchExpressions: 
                      - key: "app" 
                        operator: In 
                        values:  
                        - zookeeper-app 
                 topologyKey: "kubernetes.io/hostname" 
      terminationGracePeriodSeconds: 300 
      containers: 
      - name: k8skafka 
        imagePullPolicy: Always 
        image: gcr.io/google_containers/kubernetes-kafka:1.0-10.2.1  
        resources: 
          requests: 
            memory: "1Gi" 
            cpu: 500m 
        ports: 
        - containerPort: 9093 
          name: server 
        command: 
        - sh 
        - -c 
        - "exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \ 
          --override listeners=PLAINTEXT://:9093 \ 
          --override zookeeper.connect=zookeeper-statefulset-0.zookeeper-headless.default.svc.cluster.local:2181,zookeeper-statefulset-1.zookeeper-headless.default.svc.cluster.local:2181,zookeeper-statefulset-2.zookeeper-headless.default.svc.cluster.local:2181 \ 
          --override log.dir=/var/lib/kafka \ 
          --override auto.create.topics.enable=true \ 
          --override auto.leader.rebalance.enable=true \ 
          --override background.threads=10 \ 
          --override compression.type=producer \ 
          --override delete.topic.enable=false \ 
          --override leader.imbalance.check.interval.seconds=300 \ 
          --override leader.imbalance.per.broker.percentage=10 \ 
          --override log.flush.interval.messages=9223372036854775807 \ 
          --override log.flush.offset.checkpoint.interval.ms=60000 \ 
          --override log.flush.scheduler.interval.ms=9223372036854775807 \ 
          --override log.retention.bytes=-1 \ 
          --override log.retention.hours=168 \ 
          --override log.roll.hours=168 \ 
          --override log.roll.jitter.hours=0 \ 
          --override log.segment.bytes=1073741824 \ 
          --override log.segment.delete.delay.ms=60000 \ 
          --override message.max.bytes=1000012 \ 
          --override min.insync.replicas=1 \ 
          --override num.io.threads=8 \ 
          --override num.network.threads=3 \ 
          --override num.recovery.threads.per.data.dir=1 \ 
          --override num.replica.fetchers=1 \ 
          --override offset.metadata.max.bytes=4096 \ 
          --override offsets.commit.required.acks=-1 \ 
          --override offsets.commit.timeout.ms=5000 \ 
          --override offsets.load.buffer.size=5242880 \ 
          --override offsets.retention.check.interval.ms=600000 \ 
          --override offsets.retention.minutes=1440 \ 
          --override offsets.topic.compression.codec=0 \ 
          --override offsets.topic.num.partitions=50 \ 
          --override offsets.topic.replication.factor=3 \ 
          --override offsets.topic.segment.bytes=104857600 \ 
          --override queued.max.requests=500 \ 
          --override quota.consumer.default=9223372036854775807 \ 
          --override quota.producer.default=9223372036854775807 \ 
          --override replica.fetch.min.bytes=1 \ 
          --override replica.fetch.wait.max.ms=500 \ 
          --override replica.high.watermark.checkpoint.interval.ms=5000 \ 
          --override replica.lag.time.max.ms=10000 \ 
          --override replica.socket.receive.buffer.bytes=65536 \ 
          --override replica.socket.timeout.ms=30000 \ 
          --override request.timeout.ms=30000 \ 
          --override socket.receive.buffer.bytes=102400 \ 
          --override socket.request.max.bytes=104857600 \ 
          --override socket.send.buffer.bytes=102400 \ 
          --override unclean.leader.election.enable=true \ 
          --override zookeeper.session.timeout.ms=30000 \ 
          --override zookeeper.set.acl=false \ 
          --override broker.id.generation.enable=true \ 
          --override connections.max.idle.ms=600000 \ 
          --override controlled.shutdown.enable=true \ 
          --override controlled.shutdown.max.retries=3 \ 
          --override controlled.shutdown.retry.backoff.ms=5000 \ 
          --override controller.socket.timeout.ms=30000 \ 
          --override default.replication.factor=1 \ 
          --override fetch.purgatory.purge.interval.requests=1000 \ 
          --override group.max.session.timeout.ms=300000 \ 
          --override group.min.session.timeout.ms=6000 \ 
          --override inter.broker.protocol.version=0.10.2-IV0 \ 
          --override log.cleaner.backoff.ms=15000 \ 
          --override log.cleaner.dedupe.buffer.size=134217728 \ 
          --override log.cleaner.delete.retention.ms=86400000 \ 
          --override log.cleaner.enable=true \ 
          --override log.cleaner.io.buffer.load.factor=0.9 \ 
          --override log.cleaner.io.buffer.size=524288 \ 
          --override log.cleaner.io.max.bytes.per.second=1.7976931348623157E308 \ 
          --override log.cleaner.min.cleanable.ratio=0.5 \ 
          --override log.cleaner.min.compaction.lag.ms=0 \ 
          --override log.cleaner.threads=1 \ 
          --override log.cleanup.policy=delete \ 
          --override log.index.interval.bytes=4096 \ 
          --override log.index.size.max.bytes=10485760 \ 
          --override log.message.timestamp.difference.max.ms=9223372036854775807 \ 
          --override log.message.timestamp.type=CreateTime \ 
          --override log.preallocate=false \ 
          --override log.retention.check.interval.ms=300000 \ 
          --override max.connections.per.ip=2147483647 \ 
          --override num.partitions=1 \ 
          --override producer.purgatory.purge.interval.requests=1000 \ 
          --override replica.fetch.backoff.ms=1000 \ 
          --override replica.fetch.max.bytes=1048576 \ 
          --override replica.fetch.response.max.bytes=10485760 \ 
          --override reserved.broker.max.id=1000 " 
        env: 
        - name: KAFKA_HEAP_OPTS 
          value : "-Xmx512M -Xms512M" 
        - name: KAFKA_OPTS 
          value: "-Dlogging.level=INFO" 
        volumeMounts: 
        - name: datadir 
          mountPath: /var/lib/kafka 
        readinessProbe: 
          exec: 
            command:  
            - sh  
            - -c  
            - "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093" 
      securityContext: 
        fsGroup: 1000 
  volumeClaimTemplates: 
  - metadata: 
      name: datadir 
    spec: 
      accessModes:  
      - ReadWriteOnce 
      resources: 
        requests: 
          storage: 1Gi

动物园管理员-

apiVersion: apps/v1 
kind: StatefulSet 
metadata: 
  name: zookeeper-statefulset 
spec: 
  selector: 
    matchLabels: 
      app: zookeeper-app 
  serviceName: zookeeper-headless 
  replicas: 5 
  updateStrategy: 
    type: RollingUpdate 
  podManagementPolicy: OrderedReady 
  template: 
    metadata: 
      labels: 
        app: zookeeper-app 
    spec: 
      containers: 
      - name: kubernetes-zookeeper 
        imagePullPolicy: Always 
        image: "k8s.gcr.io/kubernetes-zookeeper:1.0-3.4.10" 
        resources: 
          requests: 
            memory: "1Gi" 
            cpu: "0.5" 
 
        ports: 
        # Expose by zookeeper-cs to the client 
        - containerPort: 2181 
          name: client 
        # Expose by zookeeper-headless to the other replicas of the set 
        - containerPort: 2888 
          name: server 
        - containerPort: 3888 
          name: leader-election 
        command: 
        - sh 
        - -c 
        - "start-zookeeper \ 
          --servers=3 \ 
          --data_dir=/var/lib/zookeeper/data \ 
          --data_log_dir=/var/lib/zookeeper/data/log \ 
          --conf_dir=/opt/zookeeper/conf \ 
          --client_port=2181 \ 
          --election_port=3888 \ 
          --server_port=2888 \ 
          --tick_time=2000 \ 
          --init_limit=10 \ 
          --sync_limit=5 \ 
          --heap=512M \ 
          --max_client_cnxns=60 \ 
          --snap_retain_count=3 \ 
          --purge_interval=12 \ 
          --max_session_timeout=40000 \ 
          --min_session_timeout=4000 \ 
          --log_level=INFO" 
        readinessProbe: 
          exec: 
            command: ["sh", "-c", "zookeeper-ready 2181"] 
          initialDelaySeconds: 60 
          timeoutSeconds: 10 
        livenessProbe: 
          exec: 
            command: 
            - sh 
            - -c 
            - "zookeeper-ready 2181" 
          initialDelaySeconds: 60 
          timeoutSeconds: 10 
        volumeMounts: 
        - name: zookeeper-volume 
          mountPath: /var/lib/zookeeper 
      securityContext: 
        fsGroup: 1000 
  volumeClaimTemplates: 
  - metadata: 
      name: zookeeper-volume 
    spec: 
      accessModes: [ "ReadWriteOnce" ] 
      resources: 
        requests: 
          storage: 1Gi

请您参考如下方法:

kafkacat -L -b kafka-statefulset-0.kafka headless.default.svc.cluster.local:9093

不起作用。在第一个 -b 标志

之后有一个空格

kafkacat -b 172.17.0.10:9093 -t second_topic -P

如果没有返回代理可解析地址的正确 advertised.listeners，这将无法工作。

kafkacat -L -b 172.17.0.10:9093

这是朝着正确方向迈出的一步，但您使用的是 Docker 网络 IP，而不是任何服务的集群/节点端口，仍然无法移植到集群外的其他机器。

installing Kafkacat inside the pod's container and trying to talk with other Kafka's broker

这是一个很好的测试，可以确保复制至少能正常工作，但不能解决外部客户端问题。

I got the right metadata but with error in the end

这可能表明您对 broker 或 zookeeper 的健康检查失败并且 pod 正在重新启动。请注意，错误从 0,2,4...开始循环

I also built headless service for talking with each Kafka's Broker through port 9093

好的，很好，但是您需要将每个代理的 advertised.listeners 设置为现在接受 9093 上的连接。

总而言之，我认为仅使用允许外部连接的现有 Kafka Helm 图表是您的最佳选择。

IT序号网 zookeeper

微信公众号号：IT虾米（左侧二维码扫一扫）欢迎添加！

Q : 823053937

热贴推荐

kubernetes之检查我的 Kafka 和 Zookeeper 功能和连接

Q : 823053937

热贴推荐

kubernetes之检查我的 Kafka 和 Zookeeper 功能和连接

kubernetes之如何在Kubernetes中处理Kafka和Zookeeper的故障情况

kubernetes之建议为Apache kafka的zookeeper和broker使用的批量方法是什么

apache-kafka之Zookeeper拒绝来自旧客户端的Kafka连接

kubernetes之Kafka无法连接到EKS上的Zookeeper集成

apache-kafka之Kafka + Zookeeper : Connection to node -1 could not be established. Broker 可能不可用

kubernetes之Kubernetes 集群中用于 Apache Drill + Zookeeper 的 Zookeeper 上的存储插件配置

chef-infra之etcd/zookeeper/consul 和其他配置管理工具(如 chef/puppet/..)之间的关系

apache-zookeeper之ZooKeeper 和 Etcd 有多好

apache-kafka之Zookeeper 在 Kafka 0.9.0 上的使用

solr之将 Zookeeper 添加到现有 Solr

Hbase 与 Zookeeper 的连接频繁连接和断开连接，但代码以 0 退出

scala之Spark -hbase-连接器 : ClusterId read in ZooKeeper is null