Kafka-Connect接收器任务忽略文件偏移量存储属性

2022-05-26 00:00:00 java apache-kafka apache-kafka-connect

我在使用融合JDBC连接器时遇到了非常奇怪的行为。我非常肯定它与融合堆栈无关,而是与Kafka-Connect框架本身有关。

因此,我将offset.storage.file.filename属性定义为默认/tmp/connect.offsets并运行我的接收器连接器。显然,我希望连接器持久化给定文件中的偏移量(它不存在于文件系统中,但它应该是自动创建的,对吗?)文档显示:

offset.storage.file.filename 要存储连接器偏移量的文件。通过将偏移量存储在磁盘上,可以在单个节点上停止和启动独立进程,并从以前停止的位置继续。

但卡夫卡的行为方式完全不同。

  1. 它检查给定文件是否存在。
  2. 如果不是,卡夫卡只是忽略它,并在卡夫卡主题中保留偏移量。
  3. 如果我手动创建给定文件,读取仍会失败(EOFException),并且偏移量将再次保留在Theme中。

这是一个错误吗?或者更有可能的是,我不知道如何使用这种配置?我了解两种保存偏移量的方法之间的差异,文件存储更适合我的需要。


解决方案

offset.storage.file.filename仅用于源连接器。它用于在输入数据源上放置书签,并记住它停止读取它的位置。创建的文件包含类似文件行号的内容(对于文件源)或表行号(对于一般的JDBC源或数据库)。

在分布式模式下运行Kafka Connect时,此文件将替换为默认命名的Kafka主题connect-offsets,应复制该主题以容错故障。

就接收器连接器而言,无论使用哪种插件或模式(独立/分布式),它们都会像任何Kafka消费者一样,将上次停止阅读输入主题的位置存储在名为__consumer_offsets的内部主题中。这允许使用诸如kafka-consumer-groups.sh命令行工具之类的传统工具来确定接收器连接器的滞后程度。

Confluent Kafka replicator尽管是源连接器,但可能是个例外,因为它从远程Kafka读取并且可能使用Kafka使用者。

我同意文档不清楚,无论连接器类型是什么(信源或信宿),此设置都是必需的,但它仅由信源连接器使用。此设计决策背后的原因是,单个Kafka Connect工作进程(我指的是单个JVM进程)可以运行多个连接器,可能既包括源连接器,也可能是宿连接器。换句话说,此设置是工作级别设置,而不是连接器设置。

相关文章