Kafka MicrosoftSqlServerSource Connect主题重复条目问题咨询

问题背景 #

需搭建Kafka MicrosoftSqlServerSource Connect捕获Azure SQL销售表的所有插入、更新事务，已在库表层级启用CDC，基于源表创建视图作为连接器输入（配置table.types = VIEW）。全量配置完成后新增/更新操作可正常同步到对应主题，但停止写入操作后，主题中最后一条消息会持续重复推送，直到新消息写入。

当前连接器配置如下：

Connector Class = MicrosoftSqlServerSource
Max Tasks = 1
kafka.auth.mode = SERVICE_ACCOUNT
kafka.service.account.id = **********
topic.prefix = ***********
connection.host = **************8
connection.port = 1433
connection.user = ***************
db.name = **************88
table.whitelist = item_status_view
timestamp.column.name = ProcessedDateTime
incrementing.column.name = SalesandRefundItemStatusID
table.types = VIEW
schema.pattern = dbo
db.timezone = Europe/London
mode = timestamp+incrementing
timestamp.initial = -1
poll.interval.ms = 10000
batch.max.rows = 1000
timestamp.delay.interval.ms = 30000
output.data.format = JSON

问题1：该现象是否为系统默认行为 #

是当前所选运行模式下的固有逻辑表现，不是连接器故障。
你配置的mode = timestamp+incrementing是基于JDBC轮询的增量同步模式，逻辑是每次轮询按记录的时间戳、自增ID拉取大于上次保存偏移量的新数据。配合你设置的timestamp.delay.interval.ms = 30000，连接器每次只会提交30秒前数据的偏移量，最近30秒写入的数据会被划入延迟窗口反复拉取，避免时间戳精度不足导致漏数。当没有新数据写入时，最后一条消息会一直留在这个30秒的延迟窗口里，被每次轮询任务捞取推送，直到新数据写入、这条消息被挤出延迟窗口，偏移量更新到对应位置才会停止重复推送。

问题2：是否存在配置遗漏导致重复条目产生 #

不是配置遗漏，是核心配置选型和你的业务需求完全不匹配：

• 模式选型错误：你需要捕获插入、更新两类事务，本质需要CDC日志读取能力，但timestamp+incrementing模式只能识别新增数据，根本无法捕获记录更新操作（更新不会改变自增ID，若时间戳列未同步更新甚至连数据变化都识别不到），和你提前开启数据库CDC的准备完全脱节。
• 输入对象选型错误：CDC能力是基于基表的事务日志实现的，视图本身不产生CDC日志，你创建视图作为同步输入的操作，本身就无法适配CDC同步逻辑。
• 现有轮询模式下缺少边界排他配置：默认的时间戳边界用>=判断，配合延迟窗口机制，直接导致边界值数据重复拉取。

问题3：该重复问题的处理方案 #

根据你的实际需求二选一即可，优先选方案二匹配最初的CDC同步目标：

方案一：保留现有轮询模式（不推荐，无法捕获更新） #

如果暂时不需要捕获更新，仅同步新增数据，调整两个配置即可解决重复问题：

• 新增配置timestamp.offset.condition = >、incrementing.offset.condition = >，将边界判断从默认的大于等于改为严格大于，避免匹配到已经同步过的边界值。
• 将timestamp.delay.interval.ms设为0，关闭延迟窗口机制，每次轮询后直接提交当前捞取到的最大偏移量。

方案二：切换到CDC模式（推荐，匹配插入+更新捕获需求） #

这是能同时满足你捕获变更、解决重复推送问题的根本方案：

1. 移除所有时间戳、自增列相关配置（timestamp.column.name、incrementing.column.name、timestamp.initial、timestamp.delay.interval.ms），将mode改为cdc，连接器会直接读取SQL Server的CDC系统表获取变更，不需要依赖字段值做轮询判断。
2. 移除视图相关配置：将table.whitelist改为实际需要同步的基表名称，table.types改为TABLE，不需要单独创建视图作为同步输入，CDC模式会自动捕获基表的插入、更新、删除操作。
3. 补充CDC必填配置：如果需要启动时同步表内历史存量数据，设置snapshot.mode = initial；如果只需要同步启动后的新增变更，设置snapshot.mode = schema_only；同时开启skip.offsets.on.no.events = true，无新变更时跳过空轮询，避免无效拉取。
4. 按需调整poll.interval.ms的值即可，不需要额外配置轮询过滤规则。

内容的提问来源于stack exchange，提问作者user19196595