同一条「业务服务 → Kafka → Kafka Connect sink → ClickHouse」链路又出了一次重复行，这次的入口在 producer：broker 有一段时间反复返回 NOT_LEADER_OR_FOLLOWER，producer 重试时把一部分记录在 topic 里存成了两份，sink 照常把两份都写进了 ClickHouse。一个一千多万行的日分区里，45 分钟的窗口多出一百多行，每个键正好两份。

算上 producer 重试，这条链路上能写出重复的入口一共四个。清理手段对它们都一样（REPLACE PARTITION runbook），能不能从源头挡住却各不相同。这篇把四个入口放在一起，再拿 sink 自带的 exactlyOnce 逐个对一遍。exactlyOnce 的结论来自 clickhouse-lab 的实验 21、23：Apache Kafka 3.7.0 加 clickhouse-kafka-connect v1.3.9（和这套系统在用的插件同一版），ClickHouse 25.3.14.14 三副本。

重复从哪个入口进来#

入口 怎么产生 见过的情形
producer 重试 请求其实写进了 broker，应答在抖动里丢了，重试又写一份，topic 里是两条 offset 不同的记录 本篇这次，一百多行
sink 插入超时重投 INSERT 超过 client 超时但服务端已经提交，框架把同一批原样再交一次 Kafka Connect 超时重投那篇里的 234 行
sink task 重启后重放 task 被重启前没来得及提交 offset，接替的 task 从上一个提交点重读，批次边界会变 lab 上造过（实验 21）
人工补发 发送超时的记录先存进重试表，事后补发时其中一部分其实早已写进去 这次补发前先按键查了 ClickHouse，没有补出新的重复

服务端的块级去重对这几种基本帮不上：重放批次的边界和原批不同，producer 重试留下的两份本来就是两条记录，只有超时重投时批次原样不变，但重投到达时第一次的记录已经被挤出了窗口（ClickHouse 的块级去重窗口）。

开着幂等的 producer 也会写出两份#

producer 用的 kafka-clients 是 3.4.0，配置里没有覆盖 acks 和 enable.idempotence。按 Confluent 的 producer 配置参考，没有冲突配置时幂等默认打开，delivery.timeout.ms 默认 120000。重复还是出现了。

AutoMQ 对 producer 的整理（2025-02）写到了这个边界：幂等只在一个 producer 会话内成立，超过 delivery.timeout.ms 报失败时，这条消息到底写没写进 broker 是不确定的。这次的样本记录正是在发出 120 秒后报发送失败、被转进重试表的。topic 里的数据因此要按至少一次对待。

exactlyOnce 在 sink 里怎么判断#

ClickHouse 官方的 Kafka Connect sink 有一个 exactlyOnce 模式，默认关闭。按设计文档，每个 topic-partition 在一张 KeeperMap 表里记下上一批的 offset 区间和一个状态：插入前写 BEFORE_PROCESSING，ClickHouse 确认之后写 AFTER_PROCESSING。新来的一批先和记录的区间比，再按状态决定怎么处理（v1.3.9 的 Processing.java 和 RangeContainer.java）：

新一批和记录区间的关系 上一批已确认（AFTER_PROCESSING） 上一批结果不明（BEFORE_PROCESSING）
终点相同，起点不晚于记录的起点 跳过 原样重插，代码注释是「Dedupe in clickhouse will fix it」，交给块级去重
整批在记录区间之后 正常写 正常写
整批落在记录区间里面 跳过 不重插，抛 State CONTAINS
越过记录区间的末尾 切掉重叠的那段，只写新的 前半段边界和记录相同才重插，不同就把前半段当 DuplicateException 交给 DLQ；后半段照写
起点更早，终点没越过记录区间 抛 State MISMATCH，打开 tolerateStateMismatch 才改成跳过 抛 ERROR State

这条链路的 sink 跑在 Confluent Cloud 上，ClickHouse 在 Aiven 上。Confluent Cloud 全托管 ClickHouse Sink 的配置页列出了 exactlyOnce（默认 false）、keeperOnCluster 和 tolerateStateMismatch，Aiven 的支持引擎列表里 KeeperMap 在 25.3、25.8、26.3 上都可用。KeeperMap 还依赖服务端配置的 keeper_map_path_prefix（KeeperMap）。按 ClickHouse 的 sink 文档，开了 exactlyOnce 就不能用 buffering。这条链路上的 sink 没开 exactlyOnce。

lab 上用真的 Connect 测了什么#

实验 21 造了两种重投：同一批原样重投，和 worker 崩溃之后批次边界变了的重放；实验 23 造了打开之后会遇到的两种卡住：State MISMATCH 和 State CONTAINS。

同一批原样重投，开不开 exactlyOnce 都一样#

实验 21 让一条 INSERT 在服务端提交之后卡在等副本确认，客户端第 30 秒超时，框架在第一次之后 53 到 54 秒用同一个 token 原样重投。三张表对照：去重窗口还在、不开 exactlyOnce 的那张，第二次被块级去重拦下；窗口设成 0 的两张都落下了第二份，其中一张开着 exactlyOnce，状态表最后记成 AFTER_PROCESSING [1, 5]。这套系统里出过的超时重投属于这一种，拦不拦得住取决于去重窗口。

批次边界变了的重放，exactlyOnce 挡得住#

实验 21 第二段在 offset 1 到 5 写进去之后立刻 kill -9 掉 Connect worker，offset 来不及提交，提交点还在 0；worker 停着的时候又来了 5 条。重启之后从 offset 0 重读，一批拿到 offset 0 到 10 共 11 条，token 和第一次不同：不开 exactlyOnce 的表去重窗口一直在，照样把 offset 1 到 5 又写了一遍；开着的表靠状态表里记的区间把重叠那段切掉，没有重复。

打开 exactlyOnce 之后，崩溃重启默认会停住#

实验 21 重启后重读的一批是 offset 0 到 10，状态表记的是崩溃前写进去的 1 到 5，正好属于前面表里「越过记录区间的末尾」那一行。更常见的时机是崩溃前、提交点之后已经写过不止一批（实验 23），这时重读的第一批起点更早、终点没越过记录区间，源码抛 State MISMATCH：

配置 结果
errors.tolerance=none（默认） 重启后约 30 秒 task 变成 FAILED；没有多写，数据都还在 Kafka，但不会自己恢复
同上，再打开 tolerateStateMismatch task RUNNING，重读的几批被跳过，没有重复
errors.tolerance=all 加 DLQ task 不停；重读的前两批（offset 1 到 10）整批进了 DLQ，而它们崩溃前已经写进 ClickHouse 了

放到这套系统的链路上，State MISMATCH 几乎每次崩溃都会碰到：一天两亿行左右、峰值每秒约 1 万条、8 个分区、offset 60 秒提交一次，两次提交之间每个分区要写一万多到几万条，一次 poll 最多 500 条，重读的第一批几乎一定落在记录区间之前。「几乎一定」是按这几个数推的。

另一种时机更麻烦：一批插入在服务端没提交、客户端只看到超时，状态停在 BEFORE_PROCESSING，之后批次边界又变了（实验里改小了 max.poll.records，这套系统的链路上对应的是重启、rebalance、升级之后一次 poll 拿到的条数不同）。源码对插入结果不明的这一段不重插：

配置 结果
errors.tolerance=none task FAILED，报 State CONTAINS；tolerateStateMismatch 管不到这一种（源码里它只用在 AFTER_PROCESSING 那一支），按文档的 Troubleshooting 删掉状态表里报错的 topic-partition 那一行再重启，删完重写，数据完整
errors.tolerance=all 加 DLQ task 不停；这一段只剩 DLQ 里那一份，ClickHouse 里从来没有过
errors.tolerance=all，没配 DLQ task 不停；这一段直接丢了，offset 照常提交

状态表存在 KeeperMap 里#

状态表本身也有风险。Aiven 的 sink 文档写明，ClickHouse 服务从备份恢复、关机或 fork 之后，connector 不保证 exactly-once（Aiven ClickHouse sink 的 Limitations）。Replicated 库的副本做恢复时，会把定义和库元数据对不上的 KeeperMap 表删掉重建；这张表如果是所在 Keeper 路径上的最后一张（比如单副本的库），Keeper 里的数据会一起清空，重建出来是空表（ClickHouse #111957，2026-07-26 提出，PR #122917 在 2026-09-30 合入修复）。25.3 的 DatabaseReplicated.cpp 里这段删表逻辑还在。

状态表、块级去重、复制元数据都在同一个 Keeper 上，开了 exactlyOnce，每一批插入前后还要各写一次状态表。这个 Keeper 出现过请求堆积：Kafka Connect 超时重投多出的 234 行全出在 3 个小时里，那段时间在途请求从常态的 5 到 15 堆到了几千。

四个入口逐条对照#

入口 exactlyOnce 能不能挡 剩下的靠什么
sink task 重启后重放 能，批次边界变了也认得出；但默认配置下重启常常停在 State MISMATCH，要打开 tolerateStateMismatch 不开的话，靠存储层按键折叠或事后清理
sink 插入超时重投 挡不住，状态停在 BEFORE_PROCESSING，原样重插，仍然靠去重窗口 调大窗口；存储层按键折叠或事后清理
producer 重试 挡不住，两份是 offset 不同的两条记录，对 connector 来说都是新数据 producer 端挡不全（幂等默认开着也出过）；落库后按键折叠或清理
人工补发 挡不住，理由同上 补发前按键查

打开 exactlyOnce 还要连着几样一起配。tolerateStateMismatch 打开之后，崩溃重启时重读的旧批次会被默默跳过，不再停下来等人；文档对它的定位是故障之后修复用，修完改回 false。errors.tolerance=all 一定要配 DLQ，并打开 errors.deadletterqueue.context.headers.enable（默认关），DLQ 记录才带上异常信息的头，比如 __connect.errors.exception.message。DLQ 不能整份重放：State MISMATCH 那些崩溃前已经写过，再写就重复；State CONTAINS 和 DuplicateException 那一段 ClickHouse 里可能没有，要补写。

这条链路最后选的是不开 exactlyOnce，接受至少一次，靠存储层按业务键折叠（ClickHouse 明细表的去重方案）。

小结#

  • 这条链路上的重复有四个入口：producer 重试、sink 插入超时重投、task 重启后重放、人工补发，清理手段都一样。
  • producer 开着幂等也会出重复：超过 delivery.timeout.ms 报失败的那条，写没写进 broker 不确定，topic 要按至少一次对待。
  • lab 上用真的 Connect 测过：exactlyOnce 挡得住批次边界变了的重放，挡不住同一批原样重投，后者开不开都要靠块级去重窗口。
  • 打开之后，崩溃重启默认会停在 State MISMATCH；插入结果不明、边界又变了的那一段不重插，只开 errors.tolerance=all 不配 DLQ 时会直接丢掉。
  • 状态存在 KeeperMap 里，和块级去重、复制元数据共用同一个 Keeper。

参考资料#

相关文章#