合流Kafka连接分布式jdbc连接器

Question

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

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

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

现在，我们已经停止了这一项，并以分布式模式启动了kafka connect服务，如下所示

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 个节点当前正在运行具有相同连接分布式属性文件的连接服务。

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 connection-*/下定义的连接器，就像您手动运行一个单独的连接器来提供属性文件的路径一样。

Answer 1

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

Answer 2

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

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

./confluent start

后记 我停止了Kafka连接

./confluent stop connect

然后我继续在两个不同的端口上手动启动定制的＜code＞connect distributed＜/code＞（＜code＞18083＜/code＜和＜code＞28083＜/code＞）

➜  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，以防您正在处理在其中一个代理中断的情况下不想丢失的东西。

Answer 3

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

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

在独立模式下，您给出的< code>N 1属性文件可以立即加载，但是对于连接分布式，您必须使用HTTP POST调用来连接REST API。

ConFluent Control Center或Landoop的Connect UI可以为这些操作提供一个很好的管理Web门户。

顺便说一句，如果您有多个代理，我将建议增加连接分布式中那些连接主题的副本因子。属性文件。

< sup>1。< sub >如果是的话，这可能是一个很好的特性，但是您必须确保在分布式模式下连接器永远不会被删除/停止，并且您最终会处于与文件系统上正在运行的内容和文件不一致的状态。