node.js - kafka Node 无法在本地环境中连接到kafka服务器

Question

我使用docker inspect命令启动了一个kafka(版本2.2.0)docker容器。

卡夫卡配置是

Using ZOOKEEPER_CONNECT=172.17.0.2:2181
Using KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.17.0.6:9092
Using KAFKA_BROKER=172.17.0.4:9092

显示网络设置

 "NetworkSettings": {
            "Ports": {
                "8778/tcp": null,
                "9092/tcp": null,
                "9779/tcp": null
            },
            "Gateway": "172.17.0.1",
            "GlobalIPv6Address": "",
            "GlobalIPv6PrefixLen": 0,
            "IPAddress": "172.17.0.6",
            "IPPrefixLen": 16,
            "IPv6Gateway": "",
            "MacAddress": "02:42:ac:11:00:06"

另外，我在kafka容器中查找了主题。

./zkCli.sh
ls /brokers/topics

主题如下所示

[connect-status, dbserver1.public.dumb_table, my-connect-offsets, __consumer_offsets, my-connect-configs]

最后，我启动了一个节点应用程序以连接到kafka，但是失败了。 (https://github.com/SOHU-Co/kafka-node#kafkaclient)

const kafka = require('kafka-node');
const bp = require('body-parser');

try {
  const Consumer = kafka.HighLevelConsumer;
  const client = new kafka.KafkaClient({kafkaHost: '172.17.0.6:9092'});
  let consumer = new kafka.Consumer(
    client,
    [{ topic: "dbserver1.public.dumb_table", partition: 0 }],
    {
      autoCommit: true,
      fetchMaxWaitMs: 1000,
      fetchMaxBytes: 1024 * 1024,
      encoding: 'utf8',
      fromOffset: false
    }
  );
  consumer.on('ready', function () {
        console.log('consumer ready');

  });
  consumer.on('message', async function(message) {
    console.log('here');
    console.log(
      'kafka-> ',
      message.value
    );
  })
  consumer.on('error', function(err) {
    console.log('error', err);
  });
}
catch(e) {
  console.log(e);
}

如果kafkaHost配置为'172.17.0.6:9092'，则会抛出错误

{ Error: connect ENETUNREACH 172.17.0.6:9092
    at Object._errnoException (util.js:992:11)
    at _exceptionWithHostPort (util.js:1014:20)
    at TCPConnectWrap.afterConnect [as oncomplete] (net.js:1186:14)
  code: 'ENETUNREACH',
  errno: 'ENETUNREACH',
  syscall: 'connect',
  address: '172.17.0.6',
  port: 9092 }

如果kafkaHost配置为'localhost:9092'，则会抛出错误

{ TimeoutError: Request timed out after 30000ms
at new TimeoutError (/Users/xisizhe/Documents/projects/game/server/node_modules/kafka-node/lib/errors/TimeoutError.js:6:9)
at Timeout.setTimeout [as _onTimeout] (/Users/xisizhe/Documents/projects/game/server/node_modules/kafka-node/lib/kafkaClient.js:491:14)
at ontimeout (timers.js:498:11)
at tryOnTimeout (timers.js:323:5)
at Timer.listOnTimeout (timers.js:290:5) message: 'Request timed out after 30000ms' }

Answer 1

if kafkaHost is configured as '172.17.0.6:9092', an error is threw out

您的Node应用程序也在Docker中运行吗？如果是这样，您将不会看到该错误。

好像您做了docker inspect，然后从中复制了一些IP，我没有看到我遇到的任何Debezium / Docker + Kafka教程中都需要的IP。

您需要主机 IP(您的Mac)；在Docker和VM网络外部进行托管(您的主机无法将请求路由到该子网)。

例如在Mac上，我检查了一个随机容器(恰好是Kafka)

docker inspect ee708cf11010 | grep IPAddress
        "SecondaryIPAddresses": null,
        "IPAddress": "",
                "IPAddress": "172.18.0.3",
                "IPAddress": "172.20.0.3",

我无法连接到那些

nc -vz 172.20.0.3 9092
# ... hangs

但是我可以连接到暴露的端口就可以了

nc -vz localhost 9092
found 0 associations
found 1 connections:
     1: flags=82<CONNECTED,PREFERRED>
    outif lo0
    src ::1 port 51180
    dst ::1 port 9092
    rank info not available
    TCP aux info available

Connection to localhost port 9092 [tcp/XmlIpcRegSvc] succeeded!

要么

使用我的主机名和nc -vz $(hostname -f) 9092

我的局域网IP nc -vz 192.168.1.105 9092

作为Debezium tutorial states，ADVERTISED_HOST_NAME必须是主机的主机名或外部IP。主机名最简单，您可以像这样将其传递给docker。

docker run -it --name kafka -p 9092:9092 -e ADVERTISED_HOST_NAME=$(hostname -f) --link zookeeper:zookeeper debezium/kafka

您的Node代码仍然可以使用localhost:9092，但是无论如何要传递正确的IP /主机名会更合乎逻辑。

除了:您不需要使用Debezium / kafka容器来使用实际的Debezium连接器，因为它只是启动了Kafka。因此，这是我的其他答案，说明如何在Docker中为在Docker中以及从主机运行的生产者/消费者应用程序正确配置Kafka

Connect to Kafka running in Docker from local machine