带有 MSSQL 的 Kafka JDBC 连接器仅流式传输 100 行

Posted 2023-03-27

【中文标题】带有 MSSQL 的 Kafka JDBC 连接器仅流式传输 100 行

【英文标题】：Kafka JDBC Connector with MSSQL only streams 100 rows

【发布时间】：2018-09-19 21:09:10

【问题描述】：

我正在使用标准的 Confluent JDBC 连接器和 Microsoft 的 JDBC 驱动程序连接到 Microsoft SQL Server 数据库。对于一个特定的表，无论我设置我的mode=timestamp 并给它一个时间戳列，或者我只是将它设置为mode=bulk，我都只能得到 100 个结果。它永远不会比这更多。以下是我的 Connect 日志末尾不断吐出的内容：

[2018-09-19 12:51:51,393] INFO WorkerSourceTaskid=MES-0 Source task finished initialization and start (org.apache.kafka.connect.runtime.WorkerSourceTask:199)
[2018-09-19 12:51:51,412] INFO After filtering the tables are: "MES2ERP"."dbo"."BatchDetails" (io.confluent.connect.jdbc.source.TableMonitorThread:175)
[2018-09-19 12:51:51,412] INFO Finished starting connectors and tasks (org.apache.kafka.connect.runtime.distributed.DistributedHerder:868)
[2018-09-19 12:51:51,475] INFO Cluster ID: GMY7vibUSHGp5xRekGDEwg (org.apache.kafka.clients.Metadata:285)
[2018-09-19 12:52:51,069] INFO WorkerSourceTaskid=MES-0 Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:397)
[2018-09-19 12:52:51,070] INFO WorkerSourceTaskid=MES-0 flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:414)
[2018-09-19 12:52:51,077] INFO WorkerSourceTaskid=MES-0 Finished commitOffsets successfully in 7 ms (org.apache.kafka.connect.runtime.WorkerSourceTask:496)
[2018-09-19 12:53:51,082] INFO WorkerSourceTaskid=MES-0 Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:397)
[2018-09-19 12:53:51,082] INFO WorkerSourceTaskid=MES-0 flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:414)

我意识到有 batch.max.rows，默认值为 100，但使用 mode=timestamp，我应该让表中的所有行都流式传输，而不是在 100 之后停止。我希望 bulk 最终也会流式传输所有内容，或者至少每个table.poll.interval.ms 重复相同的 100 个，但它没有......它只是停止。知道发生了什么吗？

哦，我的表中有数十万行。

这是我的连接器配置

name=JSON_OCN_MES-BatchDetails
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
key.converter.schemas.enable=false
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
connection.url=jdbc:sqlserver://####;databaseName=####;user=####;password=####;
table.whitelist=BatchDetails
mode=bulk
validate.non.null=false
topic.prefix=JSON_OCN_MES-

【参考方案1】：

您可以尝试更改连接器中的默认 (100) batch.max.rows 设置

"batch.max.rows": "1234",