Confluent Kafka连接分布式模式jdbc连接器答案

【问题标题】：Confluent Kafka connect distributed mode jdbc connectorConfluent Kafka连接分布式模式jdbc连接器
【发布时间】：2019-08-29 01:39:03
【问题描述】：

我们已经成功使用 MySQL - kafka 数据摄取，使用 jdbc 独立连接器，但现在在分布式模式下使用相同的连接器（作为 kafka 连接服务）时遇到问题。

用于独立连接器的命令可以正常工作 -

/usr/bin/connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-jdbc/source-quickstart-mysql.properties

现在我们已经停止了这个，并像这样以分布式模式启动了 kafka 连接服务 -

systemctl status confluent-kafka-connect
● confluent-kafka-connect.service - Apache Kafka Connect - distributed
   Loaded: loaded (/usr/lib/systemd/system/confluent-kafka-connect.service; disabled; vendor preset: disabled)
   Active: active (running) since Wed 2018-11-14 22:52:49 CET; 41min ago
     Docs: http://docs.confluent.io/
 Main PID: 130178 (java)
   CGroup: /system.slice/confluent-kafka-connect.service
           └─130178 java -Xms256M -Xmx2G -server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.a...

2 个节点当前正在使用相同的connect-distributed.properties 文件运行连接服务。

bootstrap.servers=node1IP:9092,node2IP:9092
group.id=connect-cluster
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.topic=connect-offsets
offset.storage.replication.factor=1
config.storage.topic=connect-configs
config.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
offset.flush.interval.ms=10000
plugin.path=/usr/share/java

连接服务已启动并正在运行，但它不会加载在/etc/kafka/connect-standalone.properties 下定义的连接器。

应该对服务执行什么操作，以便每当您点击命令 systemctl start confluent-kafka-connect 时，它都会运行该服务并启动 /etc/kafka-connect-*/ 下定义的连接器，就像您手动运行独立连接器时提供属性文件的路径一样。

【问题讨论】：

标签： apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

它运行服务并启动/etc/kafka-connect-*/下定义的连接器

分布式模式不是这样工作的...它不知道您要加载哪些属性文件，也不扫描那些文件夹¹

在独立模式下，您提供的N+1 属性文件会立即加载，是的，但是对于连接分布式，您必须使用HTTP POST calls to the Connect REST API。

Confluent 控制中心或 Landoop 的 Connect UI 可以为这些操作提供一个很好的管理 Web 门户。

顺便说一句，如果您有多个代理，我建议在connect-distributed.properties 文件中增加这些连接主题的副本因子。

^{1. _{如果这样做可能是一个不错的功能，但是您必须确保连接器在分布式模式下永远不会被删除/停止，并且您最终会处于与正在运行的内容和文件系统上的文件不一致的状态。}}

【讨论】：

我正在使用以下方法调用连接器： curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" 10.30.72.218:8083/connectors/ -d ' {“名称”：“linuxemp-connector”，“配置”：{“connector.class”：“io.confluent.connect.jdbc.JdbcSourceConnector”，“tasks.max”：“1”，“connection.url”： "jdbc:mysql://X.X.X.X:3306/linux_db?user=groot&password=pwd","table.whitelist":"emp","mode":"timestamp","incrementing.column.name":"empid", "topic.prefix": "mysqlconnector-" } }' 。我已经在插件路径中有连接器驱动程序，它保存在所有连接节点上。
但仍然收到此错误 - {"error_code":400,"message":"连接器配置无效并包含以下 2 个错误：\n无效值 java.sql.SQLException: 否找到合适的司机
“找不到合适的驱动程序”意味着 JAR 不是从插件路径（更具体地说是 connect-jdbc 目录）加载的。如果独立模式有效，并且您在同一台机器上运行分布式模式，那么我不确定为什么会发生这种情况
@Tony 我认为您应该确保已将 mysql jdbc 驱动程序库添加到 ${confluent.install.dir}/share/java/kafka-connect-jdbc/ 目录
@marius 随时回答stackoverflow.com/questions/53321726/…

【解决方案2】：

我可以描述我在分布式模式下启动 jdbc 连接器所做的工作：

我在本地机器上使用confluent CLI 实用程序来更快地启动服务。

./confluent start

后来我停止了 kafka-connect

./confluent stop connect

然后我继续在两个不同的端口（@987654326@ 和28083）上手动启动自定义的connect-distributed

➜  bin ./connect-distributed ../etc/kafka/connect-distributed-worker1.properties

➜  bin ./connect-distributed ../etc/kafka/connect-distributed-worker2.properties

注意：将plugin.path 设置为完整（而非相对）路径（例如：plugin.path=/full/path/to/confluent-5.0.0/share/java）

然后我可以轻松添加新的连接器

curl -s -X POST -H "Content-Type: application/json" --data @/full/path/to/confluent-5.0.0/etc/kafka-connect-jdbc/source-quickstart-sqlite.json http://localhost:18083/connectors

这应该可以解决问题。

正如cricket_007 已经指出的那样，考虑到 kafka 代理的复制因子至少为 3，以防您处理不想在其中一个代理中断的情况下丢失的东西。

【讨论】：

confluent start 将运行分布式模式...为什么你使用两个进程呢？

【解决方案3】：

分布式模式下的连接器不能像独立模式那样通过属性文件部署。改用 REST API，请参考https://docs.confluent.io/current/connect/managing/configuring.html#connect-managing-distributed-mode

【讨论】：