登录
当前位置： 首页 > 知识库问答 >
问题：

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

杨和蔼
2023-03-14

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

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

得到了-

%错误：无法获取元数据：本地：代理传输失败

我试过了-

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

 > 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和动物园管理员--在功能上的正确性？

一些关于系统的更多信息（如果需要，我可以复制所有的YAML配置，但它相当长）-

  1. 有Kafka的全集和动物园管理员的全集。
  2. Kafka的经纪人可以通过端口2181与动物园管理员对话。我的配置是-

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

我还建立了无头服务，通过端口9093与每个Kafka的经纪人交谈

Zookeeper端口：

A.2181-客户端端口

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

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

共有1个答案

劳通
2023-03-14

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

在吊舱的容器中安装Kafkacat，并试图与其他Kafka的经纪人交谈

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

我得到了正确的元数据，但最终出错了

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

总而言之，我认为只要使用现有的Kafka掌舵人图表，允许外部连接，将是您的最佳选择。

 类似资料：
  • Kafka Zookeeper连接问题

    我用的是Kafka 0.8.2-beta，有2台Ubuntu 14虚拟机: 172.30.141.127正在运行动物园管理员 172.30.141.184在经营一家Kafka经纪人 我正在启动动物园管理员实例，如果一切顺利的话。然后，我尝试启动代理并将其连接到172.30.141.127:2181。它似乎能够在特定的端口上连接并建立会话，但是由于一些似乎没有记录的异常，它失去了连接。 代理输出：

  • 使用Kerberos保护Kafka和Zookeeper

    我正在尝试在zookeeper和kafka之间设置kerberos标识。 我遵循了此处提供的配置步骤：https://docs.confluent.io/4.1.1/kafka/authentication_sasl_gssapi.html#sasl-gssapi经纪人。 Kafka代理成功连接到zookeeper集合，代理正在zNode上设置ACL。 在Zookeeper中，我可以看到znode

  • Kafka消费者连接检查

    我想在远程位置检查Kafka消费者的连接。 可以确定是否将使用者分配给分区。 在远程位置，我可以从Kafka代理获得有关该主题的详细信息。 但是消费者能否保证消费者能够收到消费者与主题分区匹配的消息？

  • “ subprocess.Popen”-检查成功和错误

    问题内容： 我想检查子流程是否成功完成执行或失败。目前，我已经提出了一个解决方案，但是我不确定它是否正确和可靠。是否保证每个进程仅将错误输出到stderr ： 注意：我对仅重定向/打印输出不感兴趣。我已经知道该怎么做。 或者： 和： 问题答案： 您需要对流程的输出做任何事情吗？ 该方法在这里可能有用。在这里查看python文档：https : //docs.python.org/2/library

  • 检查网络和Internet连接-Android

    问题内容： 我想知道下面的方法是否可以检查我是否都已连接到网络，并且实际上也可以连接到互联网。 不只是连接到不允许我访问互联网的网络吗？ 我认为确实如此，但我不确定100％。 谢谢 问题答案： 通过比较这篇文章中可接受的答案和您的代码，您正在做什么。随意比较代码。最安全的做法是在飞行模式下，关闭WiFi并在远离WiFi的位置进行一些测试，以确保安全。祝好运。

  • 有没有可能在不启用纯文本的情况下，通过SASL连接zookeeper和kafka，通过SSL连接kafka broker和它的客户端？

    我使用的是Kafka版本0.10.1。我通过SSL及其工作fine.Now连接了Kafka经纪人及其客户。我有一个查询，但有一些限制。我的限制是 不允许纯文本通信 Kafka经纪人与其客户之间的连接是SSL Kafka经纪人和动物园管理员之间通过SASL的连接（因为动物园管理员不支持SSL） 由于所有代理间通信都设置为 SSL。我有一个疑问，动物园管理员和Kafka经纪人之间的SASL连接是否可能

相关问答

Spring Boot start Zookeeper和Kafka服务器无法让docker和zookeeper连接docker内部连接云功能和App Engine我怎么能检索纬度和经度从firest和使用它在下面的功能？检查系统是否连接到Kafka

相关文章

android检查手机和无线是否连接的方法android实现主动连接和被动连接的蓝牙聊天功能去模板和功能MySQL和检查约束Modernizr检测到的功能

相关阅读

Java10 已弃用的功能和选项NumPy排序和搜索功能Java14 弃用和删除功能Kafka 和 RocketMq 的对比Kafka和Pulsar的比较

相关工具

Apache ZooKeeper定位功能功能齐全的计算器ios7自带的扫描二维码功能ActiveReports报表功能展示－ASP.NET

相关文档

ZooKeeper 分布式进程协同fbreader 和 移动 DM 和 加解密Android 开机 和 编译系统 和 binder理解 WebKit 和 ChromiumReact 和 Webpack 小书
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.