从集群外听 kubernetes 上的 kafka答案

【问题标题】：Listening to kafak on kubernetes from outside the cluster从集群外听 kubernetes 上的 kafka
【发布时间】：2019-11-25 17:55:42
【问题描述】：

我正在谷歌云平台上运行一个 kubernetes 集群，并使用 Helm 图表在其上安装了 kafka (https://hub.kubeapps.com/charts/bitnami/kafka)。我还有另一个运行 python pod 的部署。我已经使用 LoadBalancers 暴露了 kafka 和 zookeper。这是我运行kubectl get all时得到的结果，（IP地址已更改）

kubectl get all

NAME                                    READY   STATUS    RESTARTS   AGE
pod/my-kafka-0                          1/1     Running   1          3h2m
pod/my-kafka-zookeepe-0                 1/1     Running   0          3h2m
pod/my-python-6c746645f5-5xvsb          1/1     Running   0          34m

NAME                                        TYPE           CLUSTER-IP    EXTERNAL-IP     PORT(S)                                        AGE
service/kubernetes                          ClusterIP      10.10.0.1     <none>          443/TCP                                        3h16m
service/my-kafka                            LoadBalancer   10.10.0.110   35.35.135.150   9092:30769/TCP                                 3h2m
service/my-kafka-headless                   ClusterIP      None          <none>          9092/TCP                                       3h2m
service/my-kafka-zookeepe                   LoadBalancer   10.10.0.45    35.35.135.160   2181:32740/TCP,2888:31095/TCP,3888:30057/TCP   3h2m
service/my-kafka-zookeepe-headless          ClusterIP      None          <none>          2181/TCP,2888/TCP,3888/TCP                     3h2m
service/my-python                           ClusterIP      10.10.10.80   <none>          9999/TCP                                       171m

NAME                              DESIRED   CURRENT   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/my-python         1         1         1            1           136m

NAME                                         DESIRED   CURRENT   READY   AGE
replicaset.apps/my-python-6c746645f5         1         1         1       35m
replicaset.apps/my-python-848f769cd          0         0         0       136m

NAME                                        DESIRED   CURRENT   AGE
statefulset.apps/my-kafka                   1         1         3h2m
statefulset.apps/my-kafka-zookeepe          1         1         3h2m

如果我打开 python pod 的终端，我可以访问 kafka 服务。我可以使用 python 创建主题，创建生产者和消费者，并且它可以正常工作。以下是我用来测试的代码片段。

kubectl exec -it my-python-6c746645f5-5xvsb /bin/bash

from kafka.admin import KafkaAdminClient, NewTopic
admin_client = KafkaAdminClient(bootstrap_servers="my-kafka-headless.default.svc.cluster.local:9092", client_id='test')

topic_list = []
topic_list.append(NewTopic(name="test-topic", num_partitions=1, replication_factor=1))
admin_client.create_topics(new_topics=topic_list, validate_only=False)

from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='my-kafka-headless.default.svc.cluster.local:9092')

producer.send('test-topic', b'message')

from kafka import KafkaConsumer

while True:

    consumer = KafkaConsumer('test-topic',
                            bootstrap_servers='my-kafka-headless.default.svc.cluster.local:9092')
    for msg in consumer:
        print (msg)

bootstrap server的值是从kafka配置yaml文件中找到的。

- name: KAFKA_CFG_ADVERTISED_LISTENERS
  value: PLAINTEXT://$(MY_POD_NAME).my-kafka-headless.default.svc.cluster.local:$(KAFKA_PORT_NUMBER)

到目前为止，一切似乎都运行良好。现在，如果我尝试使用外部 IP 从外部访问 kafka 代理，它似乎不起作用。我可以使用以下内容查看主题。

c= kafka.KafkaConsumer(bootstrap_servers=["35.35.135.150:9092"])
c.topics()
set([test-topic'])

但是，我看不到任何使用以下内容的消息。

from kafka import KafkaConsumer

while True:

    consumer = KafkaConsumer('test-topic',
                            bootstrap_servers=["35.35.135.150:9092"])
    for msg in consumer:
        print (msg)

我也没有收到任何错误。我似乎无法弄清楚我做错了什么。

【问题讨论】：

将此属性设置为消费者并重试auto_offset_reset='earliest'
广告的侦听器仍在报告内部服务名称...您需要两个侦听器 - 一个用于内部客户端，另一个用于外部，通常在不同的端口上

标签： kubernetes apache-kafka google-kubernetes-engine kafka-python

【解决方案1】：

感谢cricket_007 的评论，我得以解决此问题。我修改了配置文件，将外部端点作为通告的侦听器包含在内，并将内部服务名称移至侦听器。

- name: KAFKA_CFG_LISTENERS
  value: PLAINTEXT://$(MY_POD_NAME).my-kafka-headless.default.svc.cluster.local:$(KAFKA_PORT_NUMBER)
- name: KAFKA_CFG_ADVERTISED_LISTENERS
  value: PLAINTEXT://35.35.135.150:9092

【讨论】：