重新启动 kafka 连接接收器和源连接器以从头开始读取

Posted 2023-03-23

【中文标题】重新启动 kafka 连接接收器和源连接器以从头开始读取

【英文标题】：Restart kafka connect sink and source connectors to read from beginning

【发布时间】：2019-08-18 00:49:17

【问题描述】：

我对此进行了很多搜索，但似乎没有一个很好的指南。

根据我的搜索，有几点需要考虑：

重置接收器连接器内部主题（状态、配置和偏移）。 源连接器偏移实现是特定于实现的。

问题：是否甚至需要重置这些主题？

正在删除消费者组。 使用不同的名称重新启动连接器（这也是一个选项），但这似乎不是正确的做法。 将消费者组重置为--reset-offsets--to-earliest 使用 REST API（它是否提供从头开始重置和读取的功能）

重新启动接收器和源连接器以从头开始读取的最佳方法是什么？

【参考方案1】：

源连接器：

独立模式：删除偏移文件 (/tmp/connect.offsets) 或更改连接器名称。 分布式模式：更改连接器的名称。

Sink Connector（两种模式）以下方法之一：

更改名称。 重置消费者组的偏移量。组的名称与连接器名称相同。

要重置偏移量，您必须先删除连接器，重置偏移量 (./bin/kafka-consumer-groups.sh --bootstrap-server :9092 --group connectorName --reset-offsets --to-earliest --execute --topic topicName)，再添加一次相同的配置

您可以查看以下问题：Reset the JDBC Kafka Connector to start pulling rows from the beginning of time?

【讨论】：

Consumer groupconnect-<connectorName> 在为接收器消费者重置偏移量时为我工作

【参考方案2】：

源连接器分布式模式 - 有另一个选项，它为偏移主题生成新消息。 例如我使用 jdbc 源连接器： 在查看偏移主题时，我看到以下内容：

./kafka-console-consumer.sh --zookeeper localhost:2181/kafka11-staging --topic kc-staging--offsets --from-beginning --property print.key=true

["referrer-family-jdbc-source","query":"query"]   "incrementing":100

现在为了重置它，我只生成另一条递增消息：0

例如：how to produce from shell with key from here

./kafka-console-producer.sh \
  --broker-list `hostname`:9092 \
  --topic kc-staging--offsets \
  --property "parse.key=true" \
  --property "key.separator=|"
["referrer-family-jdbc-source","query":"query"]|"incrementing":0

请注意，您需要执行以下操作：

删除连接器。 如上所述，生成一条带有相关偏移量的消息。 再次创建连接器。

【讨论】：

也可以只删除特定键的值，查看更多：rmoff.net/2019/08/15/…

【参考方案3】：

有点晚，但找到了另一种方法。只需将独立模式下的 offset.storage.file.name 设置为 dev/null：

#worker.properties    
offset.storage.file.filename=/dev/null

#cmdline
connect-standalone /data/config/worker.properties /data/config/connector.properties