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

标签 kubernetes apache-kafka apache-zookeeper

作为我系统的一部分，我需要在 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 配置，但是它很长)-

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

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

  1. 我还构建了 headless 服务，用于通过端口 9093 与每个 Kafka 的 Broker 对话

  2. 动物园管理员端口:

    一个。 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 图表是您的最佳选择。

关于kubernetes - 检查我的 Kafka 和 Zookeeper 功能和连接，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/60223840/

上一篇：kubernetes - Neo4J:Pod记录Neo4j无效的用户名

下一篇：kubernetes - Kubernetes 中的 Apache Flink 部署 - 可用性和可扩展性

相关文章：

kubernetes - 使用 kubectl 命令从远程 url 传递 list 文件

macos - 如何使用 xhyve 从主机中删除 Minikube 主机路径目录？

apache-zookeeper - .net 的 ZooKeeper 替代品

java - 动物园管理员 : how to correctly reconnect when session expired?

amazon-web-services - 如何从 EKS/KOPS 集群下运行的 Pod 访问 AWS？

mysql - 在 Kubernetes 上创建 mysql 容器时如何初始化？

apache-spark - 本地 Kafka 应用程序失败并显示 : NoSuchMethodError: createEphemeral

apache-kafka - Kafka以相反的顺序消费消息

java - Spring Cloud Stream 多主题事务管理

apache-zookeeper - mesos 从站未与 mesos 主站集群连接

©2024 IT工具网  联系我们