Debezium SQL Server Source Connector 设置 Kafka 代理

Posted 2023-03-10

【中文标题】Debezium SQL Server Source Connector 设置 Kafka 代理

【英文标题】：Debezium SQL Server Source Connector set Kafka broker

【发布时间】：2020-07-09 19:16:42

【问题描述】：

我正在尝试设置 Confluent Kafka 平台的 docker 环境并与 Debezium SQL Server 源连接器集成。

我遵循了this Confluent 的 Kafka 平台指南，然后遵循了this Debezium 的 SQL Server 源连接器教程。

我的代理容器被命名为broker，这个容器和其他容器（包括connect 容器）在同一个网络中，我确保它们可以相互ping&telnet。

在 Debezium 教程中，我停留在 Start The Debezium SQL Server Connector 的步骤中，因为我收到一个错误，表明连接器正在尝试通过 localhost:9092 而不是 broker:9092 访问 Kafka 代理：

[2020-03-29 10:46:30,907] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2020-03-29 10:46:32,114] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2020-03-29 10:46:32,969] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2020-03-29 10:46:34,127] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2020-03-29 10:46:35,333] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2020-03-29 10:46:36,238] WARN [Consumer clientId=server1-dbhistory, groupId=server1-dbhistory] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

直到它最终超时：

[2020-03-29 10:46:38,664] ERROR WorkerSourceTaskid=inventory-connector-0 Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
org.apache.kafka.common.errors.TimeoutException: Failed to get offsets by times in 60000ms

有趣的是，在日志的开头可以看到我的配置被成功接收（查找broker:9092）：

[2020-03-29 10:45:38,618] INFO    database.history.kafka.bootstrap.servers = broker:9092 (io.debezium.connector.common.BaseSourceTask)

...

[2020-03-29 10:45:38,655] INFO ConsumerConfig values:
        allow.auto.create.topics = true
        auto.commit.interval.ms = 5000
        auto.offset.reset = earliest
        bootstrap.servers = [broker:9092]
        check.crcs = true
        client.dns.lookup = default
        client.id = server1-dbhistory
        client.rack =
        connections.max.idle.ms = 540000
        default.api.timeout.ms = 60000
        enable.auto.commit = false
        exclude.internal.topics = true
        fetch.max.bytes = 52428800
        fetch.max.wait.ms = 500
        fetch.min.bytes = 1
        group.id = server1-dbhistory
        group.instance.id = null

这是我的配置文件：register-sqlserver.json：

 "name": "inventory-connector",
 "config": 
     "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
     "tasks.max" : "1",
     "database.server.name" : "server1",
     "database.hostname" : "sqlserver_1",
     "database.port" : "1433",
     "database.user" : "sa",
     "database.password" : "Password!",
     "database.dbname" : "testDB",
     "database.history.kafka.bootstrap.servers" : "broker:9092",
     "database.history.kafka.topic": "schema-changes.inventory"
     
   

我通过主机添加连接器如下（就像指南一样）：

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-sqlserver.json  

我展示的日志是connect 容器日志的输出。

我的完整日志中没有其他 localhost 字样，因此不用担心其他默认值 localhost 的配置可能会被我错过。

将不胜感激任何帮助:)

【问题讨论】：

每个 Docker 容器都是它自己的小 Linux 实例，所以 localhost 和 127.0.0.1 指向它自己。如果您运行docker network inspect，它应该列出每个正在运行的容器及其自己的“IPv4Address”条目，通常在 172.17.0.0/16 网络中。使用针对您的目标实例（代理）列出的 IP 地址。

嘿@AlwaysLearning，我知道它不应该是localhost 或127.0.0.1，这就是我使用broker DNS 的原因，如果容器在同一个 docker 网络中，它会自动工作，无论如何，我尝试按照您的建议使用经纪人的 IP 地址（在我的情况下是172.21.0.3）；现在我遇到了完全相同的问题——它试图到达localhost，同时将我的配置记录为bootstrap.servers =[172.21.0.3:9092]。知道为什么吗？

【参考方案1】：

问题在于广告的听众。

您正在连接到 9092 上的代理，其中每个 the config 用于将其主机宣传为 localhost 的侦听器

KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092

这意味着客户端（在本例中为 Debezium）最初将连接到您提供给它的引导服务器 (broker:9092)，但代理会将广告主机 (localhost) 交还给客户端 - 并且然后客户端将尝试连接到它。由于它们位于不同的实例上，localhost 对于 Debeziu，连接器不是代理，因此连接失败。

参考：https://rmoff.net/2018/08/02/kafka-listeners-explained/

解决方案：

使用端口29092，根据上述配置绑定到broker 广告主机，将从 Debezium 容器正确解析

"database.history.kafka.bootstrap.servers" : "broker:29092"