Kafka Jdbc Connect 时间戳+自增方式答案

【问题标题】：Kafka Jdbc Connect timestamp+incrementing modeKafka Jdbc Connect 时间戳+自增方式
【发布时间】：2019-01-08 16:21:55
【问题描述】：

我正在使用 Kafka Jdbc Connect 时间戳+递增模式将表行同步到 Kafka。参考https://docs.confluent.io/current/connect/connect-jdbc/docs/source_config_options.html#mode

挑战是表格从时间开始同步，因为默认的开始时间是 1970 年。有没有办法覆盖开始时间（即）我只想从输入给定日期的开始同步。

【问题讨论】：

标签： apache-kafka apache-kafka-connect

【解决方案1】：

如果您想从连接器的给定偏移量开始，我建议覆盖存储在 connect-offsets 主题中的信息。

通过 Kafka REST API，您可以轻松阅读本主题的内容：

http://localhost:8082/topics/connect-offsets

查看kafka-connector-jdbc的代码在你描述的用例的相关方法中：

io.confluent.connect.jdbc.source.TimestampIncrementingCriteria#extractValues
io.confluent.connect.jdbc.source.TimestampIncrementingOffset#getTimestampOffset

覆盖连接偏移主题内容似乎是目前唯一可用的方法。

【讨论】：

【解决方案2】：

您需要将timestamp.initial 设置为您想要的输入给定日期。需要设置epoch格式。

timestamp+incrementing模式下的SQL查询附加：

WHERE "DBTime" < 'system high date i.e. 9999-12-12T23:59:59+00:00' AND (("DBTime" = 'timestamp.initial' AND "DBKey" > '-1') OR "DBTime" > 'timestamp.initial') ORDER BY "DBTime","DBKey" ASC

https://docs.confluent.io/kafka-connect-jdbc/current/source-connector/source_config_options.html#mode

【讨论】：