检查我的卡夫卡和动物园管理员的功能和连接

Question

作为我的系统的一部分，我需要使用卡夫卡和动物园管理员集群之上的库伯内特斯。我使用状态集来部署它们，并为Kafka的代理使用无头服务，以便能够与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:未能获取元数据:本地:代理传输失败

我试过-

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

正如我现在所理解的，我还没有正确地配置服务，以便从集群外部与它们连接，但是我可以从集群内部连接到它们。虽然我可以在集群中与他们进行通信，但我仍然会遇到一些我无法理解的错误。

例如(将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

问题是什么？我如何检查他们之间的沟通，检查两者的功能正确性-卡夫卡和动物园管理员？

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

1. 这里有一组卡夫卡和一组动物园管理员。
2. 卡夫卡的经纪人可以通过2181港口与动物园管理员交谈。我的配置是-

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

1. 我还建立了无头服务，通过端口9093与每个卡夫卡的经纪人交谈。
2. 动物园管理员港： a. 2181 -客户端端口 b. 2888 -服务器端口 c. 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

更新-添加卡夫卡和动物园管理员配置文件-

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 

Answer 1

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，而不是任何服务的群集/节点端口，仍然不能移植到集群之外的其他计算机。

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

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

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

这可能是一个迹象，表明你对经纪人或动物园管理员的健康检查正在失败，豆荚正在重新启动。注意从0，2，4的错误循环..。

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

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

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

Answer 2

正如你在评论中提到的：

我跑了- / # nslookup headless.default.svc.cluster.local和- Server: 10.96.0.10 Address 1: 10.96.0.10 kube-dns.kube-system.svc.cluster.local nslookup: can't resolve 'headless.default.svc.cluster.local

问题与您的环境中的DNS有关，因为您的环境无法解决无头服务。DNS组件应该提供Pod的DNS名称。

你应该收到这样的信息：

/ # nslookup my-kafka-headless
Server:    10.122.0.10
Address 1: 10.122.0.10 kube-dns.kube-system.svc.cluster.local

Name:      my-kafka-headless
Address 1: 10.56.0.5 my-kafka-0.my-kafka-headless.default.svc.cluster.local

如果您想要引用Pod支持的基于DNS名称的Kubernetes状态集(在您的例子中是: headless.default.svc.cluster.local)，待遇是一种先决条件。

验证您的服务是否设置了.sepec.clusterIP: None，或者kube-system命名空间中的kube-dns-XXXX pod是否一切正常。这里您可以找到一些有关解决DNS问题的疑难解答信息。

同样，正如@cricket_007建议的那样，您可以使用舵机来部署kafka。例如，来自本源的舵图也包含HowTo。

Answer 3

我有一个类似的问题，我通过在客户端OS hostfile中添加以下更改来修复它。(在mac中，它在具有根访问权限的private/etc/host中。那么你应该sudo dscacheutil -flushcache)

127.0.0.1 my-cluster-kafka-0.my-cluster-kafka-brokers.kafka.svc
localhost my-cluster-kafka-0.my-cluster-kafka-brokers.kafka.svc

Kafkacat已经告诉你解决的地址是什么样子的：

kafka-statefulset-0.kafka-headless.default.svc.cluster.local

您还可以添加任何解析域的IP (如果它被路由到域)