Docker 中的 Kafka 无法正常工作

Posted 2023-03-10

【中文标题】Docker 中的 Kafka 无法正常工作

【英文标题】：Kafka in Docker not working

【发布时间】：2016-06-22 01:14:23

【问题描述】：

我正在尝试将wurstmeister\kafka-docker 图像与docker-compose 一起使用，但我在连接所有东西时遇到了真正的问题。

我检查的所有帖子或问题似乎都没有任何问题，但坦率地说我迷路了。 （并且在 SO 中至少有两个问题试图解决这个问题）

我认为问题在于我对docker的网络了解不足。那么问题来了：

我可以从同一个 kafka 容器消费和生产，但是，当我尝试创建另一个容器（或将我的笔记本电脑与 python 客户端一起使用）时，我遇到了几个与 advertised.host.name 参数相关的错误（在图像中这个参数为KAFKA_ADVERTISED_HOST_NAME)

我已经尝试了很多方法来设置这个变量，但它根本不起作用。

所以我正在寻找一个权威的答案（即如何自动设置这些参数及其含义）如何设置docker-compose.yml

这是我的：

zookeeper:
  image: wurstmeister/zookeeper
  ports:
    - "2181:2181"

kafka:
  image: wurstmeister/kafka
 # hostname: kafka
  ports:
    - "9092"
  links:
    - zookeeper:zk
  environment:
    KAFKA_ADVERTISED_HOST_NAME: "kafka"
    KAFKA_ADVERTISED_PORT: "9092"
    KAFKA_ZOOKEEPER_CONNECT: "zk:2181"

更新

根据@dnephin 的建议，我在以下几行中修改了start-kafka.sh：

...
if [[ -z "$KAFKA_ADVERTISED_PORT" ]]; then
    export KAFKA_ADVERTISED_PORT=$(hostname -i)
fi
...

并从docker-compose.yml 中删除KAFKA_ADVERTISED_HOST_NAME: "kafka"

我以规范的方式启动容器：

docker-compose up -d

两个容器都在运行：

$ docker-compose ps
           Name                          Command               State                     Ports                    
-----------------------------------------------------------------------------------------------------------------
infraestructura_kafka_1       start-kafka.sh                   Up      0.0.0.0:32768->9092/tcp                    
infraestructura_zookeeper_1   /opt/zookeeper/bin/zkServe ...   Up      0.0.0.0:2181->2181/tcp, 2888/tcp, 3888/tcp 

后来我做了：

docker-compose logs

一切顺利。

用于检查 IP 地址：

$ KAFKA_IP=$(docker inspect --format ' .NetworkSettings.IPAddress ' infraestructura_kafka_1)                                                                                                            
$ echo $KAFKA_IP
172.17.0.4

and

$ ZK_IP=$(docker inspect --format ' .NetworkSettings.IPAddress ' infraestructura_zookeeper_1)                                                                                                           
$ echo $ZK_IP 
172.17.0.3

然后我在两个不同的控制台中执行：

制作人：

$ docker run --rm --interactive wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092  

消费者：

$ docker run --rm --interactive  wurstmeister/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-consumer.sh --topic grillo --from-beginning --zookeeper 172.17.0.3:2181 

几乎立即，警告开始在屏幕上飞来飞去：

[2016-03-11 00:39:17,010] WARN Fetching topic metadata with correlation id 0 for topics [Set(grillo)] from broker [BrokerEndPoint(1001,ba53d4fd7595,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2016-03-11 00:39:17,013] WARN [console-consumer-79688_9dd5f575d557-1457656747003-f1ed369d-leader-finder-thread], Failed to find leader for Set([grillo,0]) (kafka.consumer.ConsumerFetcherManager$LeaderFin
derThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(grillo)] from broker [ArrayBuffer(BrokerEndPoint(1001,ba53d4fd7595,9092))] failed
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:73)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
Caused by: java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        ... 3 more

等等

在制作人的控制台中，我写了几句：

$ docker run --rm --interactive klustera/kafka /opt/kafka_2.11-0.9.0.1/bin/kafka-console-producer.sh --topic grillo --broker-list 171.17.0.4:9092                                                           
Hola
¿Cómo estáń?
¿Todo bien?

过了一会儿，我收到了这样的回复：

[2016-03-11 00:39:28,955] ERROR Error when sending message to topic grillo with key: null, value: 4 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
[2016-03-11 00:40:28,956] ERROR Error when sending message to topic grillo with key: null, value: 16 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
[2016-03-11 00:41:28,956] ERROR Error when sending message to topic grillo with key: null, value: 12 bytes with error: Failed to update metadata after 60000 ms. (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)

而在docker-compose logs

...
zookeeper_1 | 2016-03-11 00:39:07,072 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
xid:0x2 zxid:0x47 txntype:-1 reqpath:n/a Error Path:/consumers Error:KeeperErrorCode = NodeExists for /consumers
zookeeper_1 | 2016-03-11 00:39:07,243 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create c
xid:0x19 zxid:0x4b txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners/grillo Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners/grillo
zookeeper_1 | 2016-03-11 00:39:07,247 [myid:] - INFO  [ProcessThread(sid:0 cport:2181)::PrepRequestProcessor@651] - Got user-level KeeperException when processing sessionid:0x153631368b1000b type:create $xid:0x1a zxid:0x4c txntype:-1 reqpath:n/a Error Path:/consumers/console-consumer-79688/owners Error:KeeperErrorCode = NoNode for /consumers/console-consumer-79688/owners
...

更新 2

至少在docker-machine：

首先，我定义了一个名为docker-machine的变量：

DOCKER_VM=kafka_test

然后，我将docker-compose.yml修改如下：

KAFKA_ADVERTISED_HOST_NAME: "$DOCKER_MACHINE_IP"

最后在docker-machine的环境下，我执行：

DOCKER_MACHINE_IP=$(docker-machine ip $DOCKER_VM) docker-compose up -d

但在笔记本电脑上（我的意思是，不使用虚拟机，它不起作用）

【问题讨论】：

只是检查我是否没问题：当您运行一个容器时它可以工作，但是当您创建另一个容器时（当第一个容器正在运行时）它会失败？还是问题是无法连接到kafka容器？

当我尝试使用消费者连接或生产者失败时...感谢您的提问

我想您真的想将其设置为 KAFKA_ADVERTISED_HOST。此行会将 IP 地址分配给应该具有端口号的变量：'export KAFKA_ADVERTISED_PORT=$(hostname -i)'

【参考方案1】：

不是直接回答，但如果有人想了解 kafka docker 网络wurstmeister\kafka-docker，作者在network connectivity 上写了一篇很棒的 wiki。

它解释了通过 docker-compose 配置 Kafka 网络的三个主要要求。

每个代理都必须能够与 Zookeeper 通信 - 用于领导选举 等 每个代理都必须能够与其他所有代理进行通信 - 对于 复制等。 每个消费者/生产者必须能够与 每个 Broker - 用于读取/写入数据等。

几个问题：

因为只能绑定到每个唯一端口一次 单一接口，我们不能再发布 Broker 端口（9092）。 相反，我们只是公开端口。

端口：

“9092”

从 Kafka 0.9.0 开始 - 可以指定多个端口 听着。这是为了便于支持多种协议 （即 PLAINTEXT、SASL、SSL 等）和独立的内部和外部 交通。通过此更改，host.name 和 port 已被弃用 听众的青睐。 Advertisementd.host.name 和 Advertisementd.port 有 已弃用，取而代之的是 ads.listeners。

【参考方案2】：

在我的情况下，我忘记更新 docker-compose.yml environment 配置为 kafka

以前有localhost

    environment:
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://localhost:9092

更新将localhost 替换为kafka：

    environment:
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092

满：

networks:
  kafka-net:
    driver: bridge

volumes:
  kafka:
  zookeeper_data:
  zookeeper_txns:


services:
  kafka:
    image: "bitnami/kafka:2.7.0"
    networks:
      - kafka-net
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=INTERNAL://kafka:29092,EXTERNAL://kafka:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=INTERNAL
      - KAFKA_CFG_NUM_PARTITIONS=10
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=false
      - ALLOW_PLAINTEXT_LISTENER=yes
      #       10MB max message size (vs 1MB default)
      - KAFKA_CFG_MESSAGE_MAX_BYTES=10485760
      - KAFKA_CFG_REPLICA_FETCH_MAX_BYTES=10485760
      - KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
      - KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR=1
    depends_on:
      - zookeeper
  zookeeper:
    image: "bitnami/zookeeper:3.6.2"
    networks:
      - kafka-net
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

【参考方案3】：

我只是更新我的主机文件并添加：

127.0.0.1 localhost kafkaserver

它对我来说很好用。我在 Windows 10 上使用了相同的 docker 映像。

【讨论】：

这解释了如何正确地做到这一点rmoff.net/2018/08/02/kafka-listeners-explained

【参考方案4】：

这是@radek1st 答案的改进版本。

links 是旧的 docker 方式，networks 是当前的方式。

imo，进行任何形式的系统更改都是不好的，永远不需要。这也有点违背了使用 Docker 的目的。

version: '2.1'

networks:
  sb:
    driver: bridge

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    container_name: zookeeper
    hostname: zookeeper
    networks:
     - sb
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    container_name: kafka
    hostname: $KAFKA_HOSTNAME:-kafka
    depends_on:
      - zookeeper
    networks:
     - sb
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_HOST_NAME: $KAFKA_HOSTNAME:-kafka
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://$KAFKA_HOSTNAME:-kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

然后我使用以下 bash 脚本来启动。这允许我覆盖 Kafka 主机名以进行本地开发。 ./startup.sh localhost

#!/bin/bash
echo KAFKA_HOSTNAME=$1:-kafka > .env
docker-compose up -d

阅读更多 -

https://docs.docker.com/compose/networking https://docs.docker.com/compose/environment-variables/ https://docs.docker.com/compose/compose-file/#variable-substitution

【讨论】：

注意：此设置只允许网络中的其他容器到达 Kafka，而不是主机。见rmoff.net/2018/08/02/kafka-listeners-explained

【参考方案5】：

就我个人而言，我遇到了这个问题，因为 kafka 环境中缺少 KAFKA_ADVERTISED_PORT: "9092"。

 kafka:
    image : wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
      KAFKA_ADVERTISED_PORT: "9092"
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: zookeeper

【讨论】：

当您在其他 docker 容器或多个 Kafka 容器中使用 Kafka 客户端时，这将不起作用

【参考方案6】：

我用下面的代码解决这个问题：

zookeeper:
  image: wurstmeister/zookeeper
  ports:
    - "2181:2181"
kafka:
  image: wurstmeister/kafka
  ports:
    - "9092:9092"
  depends_on:
    - zookeeper
  environment:
    HOSTNAME_COMMAND: "ifconfig eth0 | grep 'inet addr' | awk ' print $$2' | awk -F: 'print $$2''"
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  volumes:
    - /var/run/docker.sock:/var/run/docker.sock

【参考方案7】：

我对这个问题的解决方案略有不同。我将 Kafka 配置为在 kafka 主机上做广告，并且因为它在 localhost:9092 上的主机上公开，所以我在 /etc/hosts 中添加一个条目以将 kafka 解析为 localhost。通过这样做，可以从其他 Docker 容器和 localhost 访问 Kafka。

docker-compose.yml：

  my-web-service:
    build: ./my-web-service
    ports:
     - "8000:8000"
    links:
     - kafka
  kafka:
    image: "wurstmeister/kafka:0.10.2.0"
    ports:
     - "9092:9092"
    hostname: kafka
    links: 
     - zookeeper
    environment:
     - KAFKA_ADVERTISED_HOST_NAME=kafka
     - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
     - KAFKA_ADVERTISED_PORT=9092
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"

更新主机文件：

more /etc/hosts
127.0.0.1       localhost kafka

【讨论】：

谢谢，对我来说KAFKA_ADVERTISED_HOST_NAME=kafka 是关键。不需要更新主机文件或同一网络中另一个撰写服务的链接以使用 kafa:9092 作为代理连接到 kafka

使用hosts文件不是正确的解决方案...rmoff.net/2018/08/02/kafka-listeners-explained

如何处理多个集群？他们必须使用不同的端口才能从主机访问。

【参考方案8】：

对于在localhost中开发应用程序，documentation中有一个解决方案：“HOSTNAME_COMMAND”

kafka:
  image: wurstmeister/kafka
  ports:
    - 9092:9092
environment:
  KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  HOSTNAME_COMMAND: "route -n | awk '/UG[ \t]/print $$2'"

希望这对其他人有所帮助...

【讨论】：

谢谢。我认为这是最便携的方式。

【参考方案9】：

只需尝试以下操作并使用服务发现，例如 this one。

zookeeper:
  image: wurstmeister/zookeeper
  ports:
    - "2181:2181"
kafka:
  build: .
  ports:
    - "9092:9092"
  links:
    - zookeeper:zk
  environment:
    KAFKA_ADVERTISED_HOST_NAME: 192.168.59.103
    KAFKA_ADVERTISED_PORT: 9092
    KAFKA_CREATE_TOPICS: "test:1:1"
  volumes:
    - /var/run/docker.sock:/var/run/docker.sock

或者你用这个：

zookeeper:
  image: wurstmeister/zookeeper
  ports: 
    - "2181"
kafka:
  build: .
  ports:
    - "9092"
  links: 
    - zookeeper:zk
  environment:
    KAFKA_ADVERTISED_HOST_NAME: 192.168.59.103
    DOCKER_HOST: 192.168.59.103:2375
  volumes:
    - /var/run/docker.sock:/var/run/docker.sock

【讨论】：

谢谢，consul.io 看起来很酷！但是，您能否更具体一点，我的意思是，您从哪里获得192.168.59.103？你能指出这个场景中的教程设置领事吗？

您阅读过文档 (github.com/wurstmeister/kafka-docker) 吗？修改 docker-compose.yml 中的 KAFKA_ADVERTISED_HOST_NAME 以匹配您的 docker 主机 IP（注意：如果要运行多个代理，请勿使用 localhost 或 127.0.0.1 作为主机 IP。）

正如我在问题中所说，如果我使用docker-machine，它可以工作，但如果我将笔记本电脑用作 docker 主机（我在 GNU /Linux 中）则不行

谢谢@user2550587 第一个和第二个有什么区别？ （我现在使用consul）

KAFKA_ADVERTISED_HOST_NAME：192.168.59.103

【参考方案10】：

我相信您用于KAFKA_ADVERTISED_HOST_NAME 的值会根据到达容器的方式而改变。

如果您尝试从另一个容器连接，使用 kafka 应该是正确的（只要您使用将该名称设置为链接别名）。

如果您尝试从主机连接，则该名称将不起作用。您需要使用容器 IP 地址，您可以使用 docker inspect 获取该地址。但是容器 IP 地址会发生变化，因此最好在容器内部使用 $(hostname -i) 进行设置以检索它。

【讨论】：

谢谢你的回答，我必须把$(hostname -i)放在哪里？在docker-compose.yml?

在入口点脚本中，或者在容器中运行的某个脚本中。