从一次搜索投诉开始

商品价格已经在管理后台改成七百七十七分，搜索页却仍显示旧值。值班者看到连接器状态正常、Kafka 有流量、消费者也没有报错，仍然不知道问题在哪里。这不是缺少一个更大的监控面板，而是系统没有把正确性拆成可以检验的契约。源表写入成功只说明 MySQL 接受了事务；连接器正常只说明任务当前在运行；消费位点前进也不能单独证明查询对象已经更新。

我们的第一个契约是身份。商品由主键 id 唯一标识，sku 是可变化属性，不能同时把它当成投影永久身份。第二个契约是内容：投影保存 id、sku、price_cents、active、revision，后续演进增加可空 note。价格用整数分，避免对账时把浮点表示误差误判成同步失败。第三个契约是删除：源表物理删除后，在线查询不能返回该商品；投影内部可以保留删除标记用于阻止旧事件复活。

把时间分成不同边界

源库事务提交、连接器读取日志、Kafka 接收记录、消费者提交投影、消费者提交位点是五个不同的时刻。把它们统称为“同步完成”会掩盖故障窗口。业务接口若需要读到刚写入的值，可以返回源库确认，或者等待一个明确的投影水位；仅等待固定两秒不能提供读己之写保证。本文实验最终停止写入并完整对账，证明该批有限变更已收敛，不能把这个结果表述成持续写入时的线性一致查询。

我们把可接受的数据延迟看成服务目标，把最终内容一致看成正确性目标。延迟高可能仍然正确，内容错误即使非常快也不能接受。例如消费者把 price_cents 当成元展示，所有消息都会成功处理，却破坏业务语义。再如删除事件被过滤，插入和修改都正常，查询副本仍保留已经不存在的商品。正确性必须覆盖操作种类和字段语义，而不是只统计消费数量。

投影不是第二个事实源

MySQL 是本书商品事实的唯一写入入口。投影数据库由同步程序维护，业务界面不能直接修改它再期待变化反向传播。否则两个存储都能产生事实，需要冲突检测、来源标识和优先级规则，这已经是双向同步问题。我们刻意维持单向链路，使重建能从源库出发得到确定结果。

这种选择也决定恢复方式：损坏的查询投影可以丢弃并重建，源库不能因为投影异常而被反向覆盖。删除标记、消费历史和检查点属于投影的恢复元数据，并不是新增的业务事实。它们可以帮助判断哪些事件已经应用，却不能单独恢复被源库与日志共同丢失的业务变更。因此可重建不等于免备份，日志保留也不等于完整灾备。

正确性检查需要覆盖什么

在实验里，对账按照主键排序读取源表全部行，读取投影里所有未删除行，比较完整 JSON 字段和值。我们同时记录行数和规范序列化后的摘要，但通过条件是实际列表相等。仅行数相同无法排除一行缺失、一行重复；仅抽样主键无法排除价格错位；仅比较最大 revision 无法排除中间对象丢失。小数据允许完整比较，生产中可以分桶与分段执行相同逻辑。

正确性还包括变更期间的行为。新字段上线时，旧对象没有收到更新事件，投影也必须有明确的默认值语义。删除后的旧事件重放不能复活对象。消费者崩溃后，重放不能重复扣款之类的业务副作用。本书投影只更新查询文档，不执行支付操作；若读者把消费者改成外部命令处理器，需要重新建立业务幂等协议，不能只复用当前表结构。

把失败语义写进接口

查询接口也应当告诉使用者数据可能落后，而不是在后台积压时继续展示毫无上下文的旧值。对于普通商品目录，可以接受短暂延迟并保留源库管理入口；对于余额、库存扣减或访问许可，延迟副本可能不足以作最终决策。同步架构应由业务风险决定读取路径，不能先搭好流式管线再让所有接口迁入。契约还应明确字段来源：哪些值来自权威表，哪些是投影计算，哪些只是展示缓存。否则对账发现差异时，团队会争论究竟该修源还是修副本。

本章练习

给一份用户目录定义同步契约，至少写出身份字段、可变字段、删除含义、可接受延迟和对账方法。再回答：用户禁用与物理删除是否相同？如果查询系统承担权限过滤，延迟期间是否允许继续展示旧结果？这两个问题的答案会直接影响后续事件模型和读取策略。

为什么从日志读取

定时查询 updated_at 看起来很简单：记住上次执行时间，下一轮读取更晚的行。这个方案在时间精度、并发提交和删除上都有边界。一个事务较早生成更新时间，却较晚提交，可能落到已经推进的时间窗口之前。物理删除的行不再存在，更无法通过普通增量查询发现。业务程序若漏写更新时间，同步程序也无法补救。

MySQL 二进制日志描述已提交变更。Debezium 的 MySQL 连接器读取日志并形成变化事件，使我们不必在每个业务写入点维护另一条发送逻辑。代价是同步系统开始依赖复制配置、日志保留、连接器位点和表结构历史。CDC 不是绕过数据库一致性的捷径，而是选择一种更接近事务提交边界的变更来源。

本实验明确要求 ROW 格式与 FULL 行镜像。行格式记录行变化，完整镜像有助于把事件中的 before 和 after 与业务对象对应。此处不是所有 MySQL 模式的比较评测；如果生产使用不同配置，需要检查连接器兼容性、数据库负载与字段可用性。不要因为实验容器能启动，就推断任意已有实例都满足接入要求。

三种位置不能混为一谈

第一种是源日志位置，例如 binlog 文件与文件内位置，也可能涉及 GTID 集合。第二种是 Kafka 的 topic、partition 和 offset。第三种是消费者组已经提交的下一个消费位置。它们关联同一条数据的旅程，却属于不同系统的命名空间。Kafka offset 为一千，既不代表 MySQL 有一千个事务，也不表示这一千条消息都已经进入投影。

连接器还维护自己的源位点与 schema 历史。源位点帮助它知道从哪里继续读取，schema 历史帮助它解释历史日志中的字段布局。只保留数据 topic，却随意删除连接器内部状态，再声称可以从原点无损续读是不成立的。重建可以使用新连接器和新 topic 重新快照，但那是另一条明确的恢复路径。

本书报告保存消息 source 中的真实文件名与位置，用来证明事件来自连接器读取的 MySQL 日志。报告只取少量样例防止体积失控，完整正确性依据是实际消费结果和全量对账。样例中的位置不能被当成整条链路的全局事务序列号，也不能跨不同源实例直接比较大小。

事务与事件的关系

一次源事务可以修改多行，连接器通常产生多条行级事件。消费者逐条更新查询投影时，读者可能短暂看到该事务的一部分变化。单个投影写入事务只保证当前处理单元原子，不会自动把所有源事务重新聚合。若业务要求多个对象一起可见，需要使用事务元数据、缓冲和边界提交，或者把必要不变量放在源库查询中。

同样，Kafka 的事务能力有适用边界。它可以约束支持事务的 Kafka 写入和消费位点组合，不能自动把任意 PostgreSQL 事务加入同一个原子提交。本文采用明确的顺序：先提交投影，再提交 broker 位点。故障窗口产生重复投递，通过幂等处理收敛。这是至少一次交付下的设计，不叫端到端恰好一次。

日志留存是一项恢复预算

消费者停止并不等于连接器停止。前者可能只在 Kafka 形成积压，后者可能导致 MySQL binlog 留存成为瓶颈。如果连接器所需日志已被清理，重新启动的任务不能凭空补出缺口。必须选择重新快照、从备份恢复或其他经过验证的补数方式。配置一天日志留存不是承诺能够容忍任何一天故障，还要计算磁盘容量、写入峰值与实际清理策略。

本实验把 binlog 过期设置为一天以便清楚展示配置边界，但没有等待一天制造日志丢失，因此不会宣称验证了日志缺口恢复。我们验证的是日志仍可读取条件下的快照与持续增量。生产演练应增加日志清理后的失败识别，并验证报警和重建决策，而不是让运维脚本盲目重试到超时。

回放需要保留解释能力

一份原始日志只有在知道当时表结构和转换规则时才容易解释。数据库新增列、连接器升级或字段编码调整，都可能改变同样字节对应的含义。恢复资料应当包含输入版本、schema历史与转换代码，而不仅是一个offset数字。日志备份和应用版本管理因此需要协作：保存消息却丢失可运行的旧消费者，会使历史回放变成临时考古；保存旧代码却没有所需日志，同样无法恢复过去状态。

本章练习

画出一条消息经过的五个提交边界，分别标记由哪个系统保存位置。假设 Kafka 已有事件而投影没有更新，列出至少三种可能原因，并说明每种原因需要读取什么证据来区分。

四个真实服务与一个消费者

随书环境使用 Docker Compose 创建 MySQL、Kafka、Debezium Connect 和 PostgreSQL。MySQL 是源，Kafka 是事件日志，Connect 运行实际 MySQL 连接器，PostgreSQL 是可查询投影。Python 程序通过真实网络连接访问它们，没有用手造 JSON 代替连接器集成。单个 broker 和单个数据分区降低环境复杂度，使我们能专注观察故障窗口。

四个服务镜像均使用真实基线取得的固定摘要，并在每次运行时保存解析后的 RepoDigests。镜像版本名称只是一项可读说明，实际摘要才说明那次运行用了什么镜像内容。程序同时执行数据库版本查询并列出连接器插件版本。若上游重新发布同名标签，读者至少可以从证据发现环境发生变化，而不是仅依靠文件名推断版本。

实验依赖固定在 lab/requirements.txt。Python 的 Kafka 客户端使用 librdkafka，数据库客户端分别连接 MySQL 和 PostgreSQL。代码没有为依赖不可用增加跳过分支；服务启动超时、没有消息或连接器失败都会使运行失败。报告中的 skipped 固定为零只是记录约定，真正的保障是每个检查条件都必须通过，失败无法进入出版步骤。

运行和清理的所有权

运行前必须设置以 cdc-book- 开头的专属 Compose 项目名。脚本只对这个项目执行 down 和卷清理，不扫描或删除机器上所有容器。固定主机端口使命令容易阅读，也意味着同一主机不能同时运行两份实验；端口冲突应当失败，不能通过杀死未知进程解决。项目名前缀不是权限隔离机制，只是防止误操作的明确约束。

export CDC_COMPOSE_PROJECT=cdc-book-learning-001
sh lab/run.sh

脚本先安装固定 Python 依赖，再启动服务，记录镜像摘要，运行实验，最后保存容器日志并清理该项目。清理放在退出处理器中，成功和失败都执行。证据目录留在宿主机，容器删除不会同时抹掉失败原因。独立下载 ZIP 后也执行同一入口，确保代码、配置和依赖完整。

为什么使用低权限连接器账户

源库初始化创建专用 cdc 账户，并赋予读取表、读取复制日志及快照所需权限。实验 root 账户只用于初始化数据、变更表结构和核对过程。这样至少把业务操作与连接器身份分开，便于理解权限用途。不过这些仍然是本地教学密码，Compose 网络也没有配置生产 TLS 或机密管理，不能直接复制到公网数据库。

连接器的 server id 与源库 server id 不同，重建连接器再使用另一个 id。复制客户端身份冲突会导致难以解释的连接行为，不能把它当作任意装饰字段。topic 前缀和 schema 历史 topic 也为在线链路与重建链路分别配置，防止测试时把两个世代的记录混进同一个数据流。

保留失败现场比美化报告更有用

证据包括运行检查列表、版本、镜像摘要、连接器配置、快照交叠记录、消费者原始输出、逐次对账和容器日志。消费者返回非零时，控制程序保留标准输出和错误输出，再让整个集成失败。失败报告不能只写一个红色状态；需要知道是连接器未运行、Kafka 没有产生消息、投影事务失败，还是结果已经不一致。

初始基线和出版流水线都在 GitHub Actions 的真实 Linux runner 上运行。本地没有 Docker 的读者仍可检查源代码、下载证据和阅读书稿，但这不等于本地已经复现实验。出版检查会把下载 ZIP 解压到另一目录，再从干净容器环境完整执行一次；这比只检查 ZIP 里有没有文件更接近读者实际使用路径。

不要把测试环境当成性能结论

两千行合成商品足以触发快照、修改、删除和重建，但不能代表千万级表的吞吐。人为放慢快照查询只用于建立可观察的交叠窗口，不是生产调优建议。单机 Docker 的网络、磁盘与资源争用也不同于独立数据库集群。我们记录完成时间方便比较运行是否异常，不用它承诺生产端到端延迟。

环境失败不能被当成业务成功

初始化SQL默认要求一个干净的独占环境，重复使用已经存在的数据库会明确失败，而不是悄悄删除未知表。读者若手工绕过入口脚本，应自行保证资源所有权与清洁状态。证据目录也不应复用旧运行内容，尤其是崩溃屏障标记；正式入口在独立环境中运行，出版包排除全部运行产物。测试失败后应先保存日志，再清理项目，不能为了让下一次变绿而把失败证据一并抹掉。

本章练习

读出一次运行的 MySQL 版本、Connect 插件版本和四个镜像摘要，说明它们各自证明了什么。然后故意占用一个实验端口，确认实验明确失败且没有清理其他项目的资源。

一个足够小但不失真的目录

实验源表 inventory.products 初始有两千行。id 是 BIGINT 主键，sku 是可显示的商品编码，price_cents 是整数价格，active 表示业务可用状态，revision 是由实验写入程序显式维护的业务修订号。这些字段刻意避免复杂序列化，让我们能够看清同步协议，而不是把大部分篇幅花在某种搜索引擎的配置上。

修订号不等于 binlog 位置。它只在实验 SQL 中递增，不是由数据库自动为所有写入保证的全局序列。连接器不能替业务程序补足漏增的 revision，消费者也不会把它当成唯一去重依据。真正的交付身份来自 Kafka 的 topic、partition、offset；对象更新顺序则在相同主键始终进入同一分区的前提下，使用分区内 offset 防止旧记录覆盖新记录。

价格为整数使完整对账有明确标准。若改用 DECIMAL，连接器可能产生带编码规则的值，客户端需要按数据类型正确还原；若改用浮点，则需要先决定业务接受的表示和比较语义。把所有数字统一转换成字符串也不是通用答案，它会改变查询排序与聚合行为。投影字段类型应当写进契约，并在真实样例上验证。

查询文档与恢复元数据分开

products_projection 使用 generation 与 id 作为复合主键。generation 标识投影世代，使在线版本和重建版本能够同时存在。document 保存业务字段，deleted 保存删除状态，source_partition 与 source_offset 保存最近一次应用的位置。恢复元数据不直接暴露给商品查询，却决定了重放与重建能否安全执行。

删除记录仍保留原对象信息，主要是为了课堂观察和阻止更旧事件覆盖。生产中这会涉及数据最小化与保留期限，需要缩减或清理敏感内容。本书商品数据是合成的，不存在个人身份信息，也没有实现合规删除工作流。不能因为查询结果隐藏了对象，就声称数据库、日志与备份已经清除了所有副本。

delivery_history 为每个消费组保存已应用的事件坐标，唯一约束把重复投递变成显式冲突。delivery_checkpoint 保存下一个已在数据库中处理的位置，和业务投影在同一 PostgreSQL 事务内写入。它有助于诊断数据库已经走到哪里，但本实现启动恢复仍使用 Kafka 提交位点；没有偷偷用数据库检查点跳过 broker 尚未确认的事件。

为什么不用一张万能事件表

把全部变化原封不动塞进一张 JSON 表可以保留信息，却不自动产生正确查询。业务仍需决定主键、删除、字段默认值、版本兼容和重建边界。我们让事件历史承担去重，让商品文档承担读取，使不同目标的约束清楚可见。历史表也不是无限增长的免费保险，它会带来索引体积和清理成本，第十六章会讨论容量边界。

同样，主键更新不能当成普通属性修改。连接器可能用删除旧键与创建新键来表达身份迁移，消费逻辑需要处理两个对象的生命周期。本实验不修改 id，因此不能宣称测试了主键迁移。sku 改变则是普通文档更新，因为 id 保持不变。设计评审时应当把“字段可修改”与“对象身份可迁移”分开讨论。

为每种操作准备可区分的标记

快照期间插入的 id 为三千零一，停机期间插入的 id 为四千零一。id 一被修改，id 二被删除；停机阶段再对一百个对象做价格递增，并删除 id 十。这样每次对账失败都能回到具体操作，不会只看到一个模糊的总数差异。标记编号没有生产含义，只为实验建立可追踪的因果关系。

源表起初没有 note。后续真实执行 ALTER TABLE 增加可空列，再修改 id 一的 note。旧对象的源库默认值为 null，在线投影需要显式完成同样的字段归一化，新世代快照则会直接读取完整结构。这个设计让 schema 演进不仅是“消息能解析”，还包含查询文档语义的一致性。

先写下不变量

我们要求同一世代每个 id 最多一条投影记录，删除对象不进入查询结果，已应用的同组事件不能再次修改内容，同一分区内旧 offset 不能覆盖新 offset，投影与数据库检查点在事务提交时共同可见。每个不变量分别由主键、查询条件、历史唯一约束、条件更新和事务实现。

不变量也有边界：offset 不能跨分区比较，世代不能随意共用消费组历史，连接器更换 topic 后不能沿用旧 topic 的数值水位。实验用新的世代和新的消费组完成重建，就是为了避免把这些命名空间压扁成一个整数。

本章练习

为商品增加多租户字段，说明唯一主键、Kafka 键、投影主键和对账排序需要怎样变化。再指出：仅把 tenant_id 放入 document，而仍然用全局 id 作为键，会在哪些情况下覆盖另一租户的数据。

第一次接入为什么需要快照

数据库已经存在两千行，连接器此时才开始读取日志。如果只订阅未来变化，长期未更新的商品永远不会进入查询系统。初始快照解决这个起点问题：读取一份已有数据，并把后续日志接到这份数据之后。真正困难的地方是读取期间业务不能长期停止，快照与日志必须围绕一致性边界协调。

Debezium MySQL 初始快照会建立相应的读取和日志位置关系，具体锁与事务流程受版本及配置影响。本书使用连接器内置 initial 模式，没有自行实现数据库复制协议。我们验证该版本在当前权限、ROW 日志与单表条件下的最终结果，不把一段教程伪代码当成所有版本的完整内部实现。

制造一个可以证明的交叠窗口

普通两千行查询可能在几十毫秒内完成，后台同时启动写入并不能证明它真的与快照交叠。随书连接器为 products 指定快照查询覆盖，在 WHERE 中调用 SLEEP，使每行读取有小幅延迟。这个查询只用于实验，不推荐在生产表上照抄，也不依赖它提升数据一致性。

控制程序轮询源库 SHOW FULL PROCESSLIST，直到看到连接器正在执行带 SLEEP 的真实 SELECT。它记录查询身份和源库 UTC 时间，然后在独立业务连接中插入一行、更新一行、删除一行。写入结束后再次查看相同查询仍然活跃。前后观测与时间戳保存为 snapshot-write-overlap.json，检查要求两个观测都存在。

这份证据说明写入发生在实际快照扫描仍然执行的时间窗口内，而不只是发生在任务名为“snapshot”的步骤附近。查询已开始，连接器读取事务的快照边界已经由其内置流程建立；实验不声称从 processlist 可以直接读取 InnoDB 内部 read view 的全部内容。最终的完整对账才负责验证快照和增量衔接没有遗漏这些变化。

观察读事件与日志事件

连接器快照事件的 op 为 r，持续日志中的创建、修改和删除分别表现为 c、u、d。集成要求四种操作都实际出现，不能只看到最终行数相同就宣布验证了所有路径。数据消息中的 source 还包含真实 binlog 文件和位置，我们保存样例以确认链路读取的是真实日志。

假设 id 二在快照中被读到，稍后又被业务删除，消费者最终需要把它标记删除。假设 id 一的快照值还是旧价格，随后更新事件会覆盖成新价格。插入 id 三千零一可能通过日志进入数据流。连接器内部具体事件时序必须以实际记录为准，我们不要求从某个偶然排序推导全局序列。

快照完成不等于全链路完成

连接器结束表扫描后，Kafka 可能仍有待投递记录，消费者也可能尚未写完投影。因此不能拿快照完成日志直接向业务宣布查询已经可切换。我们等待消费者读取并提交，随后在源库写入停止的条件下逐行比较全部字段。只有比较相等，才能说该批有限数据已经收敛。

本实验消费者通过消息空闲窗口结束一次有限运行，这是一种测试驱动方式，不是生产流处理生命周期。若机器变慢导致事件晚到，对账会失败而不是把空闲当成正确。生产消费者通常持续运行，并用明确的水位和业务探针判断追平程度；不能把三秒无消息当成永久没有新变化。

快照负载与生产选择

全表读取会消耗数据库 IO、缓存和网络，也可能延长事务存活时间。对大表安排快照窗口时，要评估并发写入、日志保留与磁盘余量，而不是只调高 fetch size。增量快照、只读副本接入或分批初始化都是可讨论方案，但各有一致性与版本要求，本实验没有把它们作为已验证功能。

快照查询覆盖还会改变选取范围，错误过滤条件可能永久漏掉对象。如果生产必须过滤租户或软删除行，应明确快照过滤与持续变更过滤是否具有相同语义，否则初始数据和后续数据会来自两个不同集合。本书读取全部商品，包括 active 为零的对象，把是否展示留给查询契约。

本章练习

删除人为 SLEEP 后再运行实验，解释为什么交叠检查可能失败，且这个失败不能证明连接器快照算法错误。再设计一种不依赖固定等待秒数的替代同步点，写出它能够证明的边界。

一个位置需要带上身份

“当前位点是一千”几乎没有诊断价值。至少应当知道它属于哪个源实例、哪个日志文件，或者哪个 topic 的哪个分区。消费者的 next_offset 又表示下一条待读位置，和最后应用的记录 offset 差一。对这些概念没有统一约定，最常见的结果就是恢复时跳过一条或反复重放一条。

本书的投影行保存 source_partition 与 source_offset，但这里的 source 指 Kafka 交付来源，并不是 MySQL binlog 原始坐标。字段命名只是局部实现，阅读代码时必须看清它存的实际值。MySQL 位置保留在 Debezium 消息的 source 中，不能与这两个整数混用。生产中更好的命名可以明确使用 delivery_partition 和 delivery_offset，减少误解。

从快照集合到变化日志

快照给出一个受一致性边界约束的数据集合，日志描述其前后的变更。正确衔接不是简单地“先查全表，再从当前日志末尾开始”，因为两步之间可能已经发生业务提交。连接器负责建立快照与日志位置的关系，我们负责保留其状态、消费结果并验证最终集合。

如果自行编写初始化程序，先读表再查询日志位置，读取期间的更新可能丢失；先记录位置再读表，又需要处理重复和快照读到新值后旧日志回放的问题。两种顺序都需要严密协议，不能仅靠一个时间戳补洞。使用成熟连接器并不免除理解这些边界，只是避免在教学实验里重新实现复制算法。

为什么分区顺序只解决一部分问题

Kafka 只保证分区内顺序。我们为数据 topic 使用一个分区，并保持连接器产生的主键作为记录键。消费者依次处理，因此同一商品的快照、修改和删除具有可用的交付顺序。投影再加上 offset 条件，防止显式重放旧记录造成倒退。

增加分区后，同一主键仍需稳定路由，但分区数变化可能使之后的键落到不同分区。本书条件更新只接受同一分区中的更高 offset，因此跨分区迁移不是自动支持的在线操作。直接把分区数调大可能使对象停止更新或产生顺序歧义。生产迁移可以建立新 topic 和新世代重建，再以明确切换协议替换旧流。

多个源表还带来关联问题。商品表与分类表分别更新时，分区顺序不能自动给出跨表一致视图。需要选择反范式事件、事务边界聚合、查询时关联或者接受暂时不一致。本书只验证单表投影，不把单表结果延伸成跨表事务保证。

什么时候可以说追平

在持续写入的系统里，读取到当前 Kafka 末尾只能说明赶上某次观察到的末尾，新消息仍可能正在源库或连接器缓冲中。业务验收最好使用带身份的探针：在源库提交一个可识别变更，等待投影观察到同一业务修订，再判断相关读取路径。全局追平需要更完整的水位协议，不能只看 lag 等于零。

实验使用更强但范围更小的条件：一批源写入结束，连接器和消费者处理后，对整个有限集合逐行比较。对账时没有其他业务写入，因此源表与投影有共同稳定参照。这个条件适合验证正确性，不能替代生产持续变化条件下的延迟服务目标。

位点持久化与恢复动作要成对

连接器位点存储在 Kafka Connect 内部 topic 中，消费者位点保存在消费组状态中，数据库检查点保存在 PostgreSQL。恢复手册必须明确哪一个是恢复依据，哪一个只是诊断证据。本书消费者从 Kafka 组位点恢复，数据库历史表确保重投安全；不会因为数据库检查点更大就直接跨过未确认消息。

如果消费者位点被手工推进而投影没有对应内容，程序可能永远看不到被跳过的记录。若位点回退，则会产生更多重复，但当前幂等设计能够处理已保留的历史范围。两个动作风险不对称，恢复时应优先保留原始位点和证据，再选择回放或重建，避免用一次不可追踪的 reset 掩盖问题。

不要用墙上时钟替代日志顺序

两个消息的事件时间可能相同，机器时钟也可能调整。即使时间戳精确到微秒，也不自动具有分区内交付位置的唯一性。业务修订号可以表达某个对象的变化先后，但前提是源应用真正维护它。当前实现选择真实Kafka坐标处理交付身份，选择完整源行定义业务内容，各自承担能够证明的范围。把所有顺序都压成updated_at，通常只是把复杂性藏到下一次边界故障里。

本章练习

给定两个分区分别位于 offset 十和二十，说明为什么不能声称全局水位为二十。再列出把分区数从一改为三之前必须验证的键路由、旧数据、投影条件和切换步骤。

先看真正到达客户端的字节

连接器文档展示的逻辑事件，与 Kafka 客户端收到的 JSON 外形可能不同。Kafka Connect 的 converter 可以附带 schema 和 payload，也可以只输出 payload。消费者若默认 op 一定在顶层，却收到包含 schema 的外层对象，就会在真实集成中失败。这个错误无法靠手写一份符合预期的 JSON 单元测试发现。

本实验在连接器配置中明确设置 key 与 value 的 JsonConverter，并关闭 converter 的 schema 包装。此处关闭的是 JSON 外层携带的 Connect schema，不是关闭 MySQL 表结构变化处理，也不是说事件没有类型语义。连接器仍需维护自己的表结构历史；消费者仍需理解字段类型和可空性。

我们在早期真实运行中遇到过包装层不一致，随后改为显式连接器级配置。这个经历说明配置需要与收到的真实消息一起验收。仅在 Compose 环境变量中写一个相似名称，不能证明运行时属性一定生效。最终实验直接通过真实消息进入投影，错误形状会在解析或字段检查时失败。

before、after 与操作类型

读快照和创建事件有可用的 after，更新事件同时可能包含 before 与 after，删除事件的 after 为 null。消费者先检查 op 只属于 r、c、u、d，再根据操作选择业务行。删除时保留 before 的对象身份与字段，设置 deleted；普通更新则保存 after 的完整业务内容。

这是一种完整文档投影方式。我们不把更新事件当成字段补丁，也不尝试通过两次 after 相减推导业务命令。原因是数据变化不等于业务意图：价格减少可能来自促销、纠错或补偿，单凭行镜像不能决定是否给用户发券。CDC 消费者适合维护查询事实，不能随意把每次数据库更新解释为一次外部动作。

键和值必须相互校验

Kafka 记录键包含商品 id，值里的业务行也包含 id。消费者要求二者一致，再写入投影。这个断言可以暴露不符合当前契约的转换配置或错误消息，而不是静默把一个对象写到另一个对象的位置。我们没有接收任意 topic 的通用入口，启动绑定还要求世代、消费组与 topic 关系保持固定。

键稳定性决定顺序与压缩行为。如果把变化中的 sku 当键，修改 sku 可能让同一对象进入不同位置，旧键对应的查询记录也可能残留。对于组合主键，需要结构化编码避免字符串拼接歧义；例如租户一加对象二十三，与租户十二加对象三，不能因为缺少分隔协议而生成相同字节。当前单字段 BIGINT 不存在这个组合歧义。

删除包络与 tombstone 的职责

Debezium 删除通常先产生带 d 操作的业务包络，随后产生相同键、空值的 tombstone，用于 Kafka 日志压缩语义。消费者不能把空值当成格式错误，也不能只处理 tombstone 而忽略前面的删除信息。随书程序让 tombstone 参与交付历史和检查点更新，但不把它再次转换成业务行。

投影内部删除标记与 Kafka tombstone 是两个概念。前者防止旧业务记录重放后复活对象，后者服务于 broker 对键的压缩清理。Kafka 清理完成之前，旧创建事件仍可能存在；清理之后，也不能把压缩 topic 当成永远保存每次业务操作的审计账本。日志策略应当根据恢复需要明确设计。

事件身份与业务身份分工

同一商品会对应许多事件，因此商品 id 不能作为唯一交付身份。相反，同一个 Kafka 坐标可能被多次投递，它仍然代表同一条记录。delivery_history 使用消费组、topic、partition、offset 联合作为键，使重启和显式回放能够识别重复。投影主键则使用世代与商品 id，使多次变化聚合成当前查询状态。

这种区分也防止把重建做错。新世代需要读取完整数据，不能被旧消费组的历史去重挡掉。程序建立一对一绑定，禁止同一消费组被随意用于另一个世代，也禁止同一世代混入另一条 topic。真正需要新投影时，创建新的组和新的世代，并从明确的输入开始。

本章练习

分别写出删除包络、tombstone 和普通更新的处理结果，不需要伪造集成报告。然后修改 converter 的 schema 设置，预测当前消费者会在哪一步失败，并解释为什么应修改契约或解析器，而不是直接忽略所有缺少 op 的消息。

重复并不是例外路径

只要投影数据库和 Kafka 位点不处于同一个原子事务中，就存在数据库成功、位点尚未确认的窗口。进程在窗口内退出，重启后 broker 会再次投递该记录。把重复视为罕见脏数据，然后依靠人工删除，会让正常恢复变成危险操作。我们先承认至少一次交付，再让重复处理保持业务结果不变。

幂等的对象必须具体。把同一商品文档设置成相同值，通常可以幂等；把价格每次加一则不可以。随书消费者使用事件 after 的完整值覆盖投影，并使用历史唯一约束阻止同组坐标再次执行更新。若以后需要累计统计，应当保留事件级去重并在同一事务内更新计数，不能只依靠文档 upsert 的经验。

一次事务包含三件事

处理记录时，程序开启 PostgreSQL 事务，先向 delivery_history 插入事件身份。插入成功才应用业务变化；若唯一约束冲突，说明该消费组已经处理过该记录，跳过业务更新。随后推进数据库中的 delivery_checkpoint，最后共同提交。历史记录、查询文档和数据库检查点的可见性属于同一个数据库事务。

如果更新投影失败，事务回滚，历史插入也回滚，下一次投递仍能正常尝试。如果先单独提交历史，再更新投影，进程可能在中间退出，重启时看到历史存在而永远不再补投影。幂等表的存在本身不提供保证，关键在于它与实际状态变化共享提交边界。

数据库事务提交以后，程序再同步提交 Kafka 位点。此时第二个系统的操作失败会产生重复，不会让尚未持久化的投影被提前确认。反过来先提交位点再写数据库，会在中间崩溃时丢失投影更新，因为消费组已越过该记录。顺序决定主要故障结果：我们选择可重试的重复，避免无证据的跳过。

offset 条件是第二层防线

历史去重适用于同一消费组的重复坐标。显式回放或不同恢复场景还需要防止旧事件把新文档覆盖。投影更新条件要求分区相同且新 offset 更大。因此即使尝试重放旧创建事件，已经应用的删除标记不会倒退成未删除状态。

这个条件依赖固定 topic 和键路由，所以程序启动时验证绑定关系。新 topic 的 offset 从零开始，不能直接与旧 topic 的大数比较；换分区后的数字也没有大小关系。把这些限定写成数据库绑定和启动拒绝，比只在 README 中提醒使用者更可靠。错误组合会在消费之前退出。

真实重复实验怎样验收

首轮消费结束后，控制程序用同一消费组、同一世代显式从分区起点重放。这个操作只用于实验，生产中不能随意改变运行组的位点。消费者报告重复数等于首轮消息数，新鲜处理数为零；随后再次读取源库与投影，要求全部行与字段仍然相等。

我们还测试三种错误组合：旧消费组用于新世代、同一世代换消费组、同一世代换 topic。每次启动都必须明确报绑定错误，投影行数、历史行数和绑定数量保持不变。这样证明参数拒绝发生在数据处理前，而不是等错误结果出现再靠对账发现。

幂等历史也有生命周期

永久保存每个事件坐标会增长存储，删除历史又可能使非常旧的重复再次进入业务逻辑。对纯当前状态投影，offset 条件仍提供一部分保护，但涉及累加、外部副作用或跨对象操作时，不能随意缩短去重窗口。清理策略应与可回放范围、日志保留和重建方案一起设计。

本实验不实现历史清理，因为有限数据允许完整保留。它也不实现跨系统支付幂等或外部通知去重。若消费者要调用远程服务，需要远端接受稳定业务幂等键，并明确未知结果查询协议；本地历史事务不能原子包含一个任意 HTTP 调用。

本章练习

把投影更新想象成一次积分加一，说明只用 ON CONFLICT 更新对象为什么不够。然后画出“历史已提交、投影未提交”和“投影已提交、位点未提交”两种故障窗口，解释哪一种被当前事务边界消除了。

把故障放在明确的提交窗口

测试消费者恢复时，随便启动进程再立刻 kill，可能只杀在初始化阶段。这样的绿色测试不能证明最危险的窗口已经发生。我们为实验增加一个受控屏障：消费者成功提交一条 PostgreSQL 事务后，尚未调用 Kafka commit，写出包含 topic、partition、offset 的标记文件，然后阻塞等待。

控制程序不是根据固定睡眠猜测进程走到哪里，而是等待标记出现。它进一步查询 PostgreSQL，要求 crash 消费组的交付历史恰好一条；再通过 Kafka 客户端查询该组实际提交位点，要求未提交，或下一个位置尚未越过当前记录。两个存储的观测共同确认故障窗口真实存在。

此处标记文件只负责实验同步，不是生产检查点，也不参与消费者恢复。生产逻辑仍以 broker 位点和数据库历史为依据。把测试控制面与运行时状态分开，才能避免因为测试注入了一份额外真相，程序才偶然恢复成功。

SIGKILL 为什么有意义

控制程序向真实子进程发送 SIGKILL，并检查退出码对应该信号。进程无法执行 finally 或优雅关闭逻辑，因此不会在退出时补交 Kafka 位点。这个实验比在函数中抛异常更接近突然进程消失，但仍然不等价于数据库断电、磁盘损坏或主机内核故障。数据库服务和 broker 在整个实验中继续运行。

当前实现选择新的 crash 消费组和 crash-gen 世代，从数据流起点读取。这样首条事件一定能触发故障窗口，也不会污染在线 live 投影。控制程序确认数据库已经持久化一条事件后杀进程，再以同样组和世代重启普通消费者。重启必须至少报告一次重复，然后完成剩余数据。

预期结果是重复，不是丢失

Kafka 组位点尚未确认，所以首条记录会再次投递。delivery_history 已有同组同坐标，消费者识别重复，不再应用业务变化；数据库检查点仍可推进，随后位点提交成功。最终 crash-gen 投影与源表逐行相等，这同时检查重复记录和后续记录没有被错误跳过。

如果恢复后没有重复，可能是测试屏障位置错了，也可能是位点被自动提交。我们显式关闭 enable.auto.commit，并关闭自动 offset store，提交由业务事务成功后的代码控制。测试不仅检查最终状态，还检查 broker 位点与重复计数，就是为了避免一种错误实现恰好得到相同结果却没有经过预期路径。

同步提交与吞吐取舍

每条记录都执行数据库事务和同步 Kafka commit，吞吐会受到往返延迟限制。这里优先让故障顺序易于观察。生产可以把同分区连续记录批量放进一个数据库事务，再提交批次末尾位点，但必须确保批次内全部记录都已持久化，失败时不能越过未完成对象。

并发消费者还需要处理分区撤销。若一个线程仍在写旧分区，另一个线程已经提交该分区的新位置，就可能违反预期顺序。当前教学程序单线程轮询与处理，不声称验证了多线程 rebalance 协议。扩展并发前应补充真实分区转移和处理中撤销的故障实验。

恢复证据应该能回答什么

crash 标记说明被杀时处理了哪条记录；数据库查询说明事务已经提交；broker 查询说明位点尚未确认；进程退出码说明没有优雅收尾；重启报告说明发生了重复；完整对账说明最终内容收敛。这些证据各自覆盖不同断言，不能用一张“测试通过”截图代替。

还有一个常见误区：把下一个位点与最后处理位点直接相等比较。Kafka commit(message=msg) 提交的是这条消息之后的位置，因此记录 offset 为零时，成功确认通常意味着 next offset 为一。故障断言使用“没有越过当前记录”的关系，避免在边界处把已提交误判为未提交。

本章练习

把屏障移动到 Kafka commit 之后，预测 broker 位点、重启重复数和最终对账会怎样变化。再说明为什么两种实验都可能最终相等，但只有当前窗口能验证数据库已提交而 broker 尚未确认的重复恢复路径。

先区分是谁停止了

业务源库继续写入时，消费者可以停止，连接器也可以停止，broker 还可能不可用。三者形成的积压位置不同。消费者停止而连接器正常，变化主要积压在 Kafka；连接器停止而数据库继续写入，需要依赖 MySQL 日志仍然保留；broker 不可用则可能让连接器无法交付，同时增加源日志追赶压力。一个统一的“同步延迟”数字无法解释这些不同恢复路径。

本章真实实验只停止消费者，数据库、Connect 和 Kafka 继续运行。控制程序在没有消费者进程存活的阶段，对一百个对象执行价格递增与 revision 更新，再删除一个对象并插入另一个对象。它先比较源表与 live 投影，要求两者确实不同，证明查询副本在停机期间保持旧状态。随后才重启消费者读取积压。

为什么停机前后都要检查

如果只启动消费者并检查最终相等，测试可能因为没有成功产生写入而轻易通过。我们增加“停机时可观察到旧数据”的断言，再要求恢复消费报告包含至少一百条更新和一条删除，最后执行完整对账。这三步分别证明故障造成影响、真实变化进入流以及结果已经修复。

这里的一百条更新来自一百次真实源库提交，连接器从 binlog 捕获。它们不是直接塞入 Kafka 的测试消息，也不是在 PostgreSQL 中手工改好结果。这样消费者停机与追平路径经过了全部真实组件，能够暴露连接器过滤、消息形状和投影逻辑的问题。

用流量关系解释追平

设源数据变化进入 Kafka 的平均速率为每秒一百条，消费者恢复后只能处理每秒八十条，即使所有服务都显示运行中，积压也会继续增加。只有恢复处理能力大于持续输入，积压才会下降。追平时间粗略取决于积压量除以净处理能力，但峰值、批处理和数据库锁会让真实曲线偏离简单估算。

因此恢复容量要预留余量。平时刚好满负荷的消费者，一旦停机就没有追赶空间。增加实例也不一定有效：单个分区最多由一个组成员消费，本书只有一个数据分区，增加同组进程不能把该分区平均切成多份。容量设计必须同时考虑分区、键顺序、数据库写入与位点提交成本。

lag 是交付指标，不是内容正确性

Kafka lag 通常比较日志末尾与消费组提交位置。消费者提前提交位点会让 lag 看起来很低，投影却可能缺行；消费者事务很慢，lag 较高也可能只是延迟。连接器尚未把源日志写入 Kafka 时，消费组 lag 甚至可能为零，而源库仍有未交付变化。因此 lag 适合定位交付进度，需要与源探针和对账配合使用。

本实验报告消费消息数、操作种类和完整内容比较，没有把一轮总耗时当成稳定性能基准。生产可增加源提交时间、连接器时间与投影可见时间的差值分布，观察长尾而不是只看平均值。时间字段还会受到机器时钟偏差影响，应结合事件身份与日志位置分析。

恢复过程中不要扩大故障

积压追赶会突然提高投影写负载，可能挤压线上查询。如果让消费者无限并发，数据库连接、事务日志与磁盘都可能成为新瓶颈。更合理的做法是限制批量、连接数和最大写速率，观察查询延迟后逐步提速。限速使追平更慢，但能够保住主要业务，取舍需要写入恢复目标。

如果 Kafka 留存不足，恢复时所需事件可能已经被清理。auto.offset.reset=earliest 只会从仍然存在的最早记录开始，不能恢复已消失的历史。生产应当检测位点越界并明确选择重建，而不是把“成功读到一些消息”误认为“缺口已经补齐”。本实验在短时停机内完成，没有模拟 broker 数据过期，相关策略属于设计要求。

结束恢复需要业务验收

当积压接近零，还应检查删除对象是否消失、新插入对象是否存在、关键字段是否正确。我们的逐行核对覆盖所有商品，生产可以先运行重要对象探针，再分段进行全量对账。只有交付进度和内容检查都满足要求，才能结束恢复事件，避免下一次用户查询才发现残留错误。

重试应当有可观察的代价

投影暂时不可用时重试是合理动作，但重试速率需要受控。无限快速重连会消耗数据库连接和日志，也会挤占同机其他业务。生产客户端应区分暂时故障与契约错误，使用退避和告警；解析不支持事件不应每秒重启一次而永远不通知维护者。本书有限集成遇错即失败，便于留下清楚现场，尚未实现长期运行的退避控制器。扩展前应保留这种错误可见性。

本章练习

假设停机一小时产生三十六万条积压，持续输入每秒一百条，恢复能力每秒三百条，估算理想追平时间。随后列出至少四个会让实际时间更长的条件，并说明哪些条件可以通过增加消费者实例解决。

删除比更新更容易被忽视

插入和更新都有 after，删除的 after 却为空。一个只把非空 after 写入查询库的消费者，看起来可以正常同步大部分业务，却永远不删除旧对象。用户常常先在搜索结果里发现下架内容，团队才意识到同步契约缺了一个操作。删除必须从消息模型、存储状态与查询过滤三处共同设计。

本实验消费 d 包络后保存 deleted=true，查询投影时固定排除删除记录。快照期间删除 id 二，停机期间删除 id 十，两个阶段最终都完整对账。对账列表来自源表实际仍存在的行，因此任何未被隐藏的残留商品都会使比较失败。我们没有只统计删除事件次数，而是检查其业务可见结果。

为什么保留删除水位

假设商品先创建、再更新、最后删除。如果恢复程序从较早位置回放创建事件，而查询库已经物理删除对应记录，简单 upsert 可能把它重新插入。保留删除记录及最后 offset，可以在重放时拒绝旧事件。这是查询状态的防倒退策略，不是把删除变成永远保存业务内容的理由。

生产若要清理删除标记，需要确认旧事件不会再合法回放到当前世代，或者使用独立持久水位阻止复活。可以在重建后整体回收旧世代，让新世代从当前源事实出发；不能在缺少恢复边界时单独删除所有 tombstone，再继续使用任意旧 offset 重放。清理与回放必须有共同协议。

软删除与物理删除是不同事件

active=0 仍然是一行数据。本书投影保留它，并让业务查询决定是否展示。若把 active=0 当成删除，需要在快照、更新和重建三条路径采用相同规则；否则初始快照可能过滤掉它，之后一次普通更新又把它写回来。源库字段语义必须先于消费者条件判断确定。

物理删除则意味着源表不再有该 id。连接器 d 事件负责把变化传到投影，后续新快照不会再包含它。若上游执行表截断或批量替换，事件类型和连接器支持可能不同，不能直接套用逐行 DELETE 的实验结论。本书没有测试 TRUNCATE，生产接入时应明确禁止或单独验证这类操作。

Kafka tombstone 不会清理所有副本

记录空值可以影响压缩 topic 的旧键保留，但 broker 清理是异步策略，其他存储和备份也不会因此自动删除。删除包络、查询隐藏、日志清理、缓存失效和备份保留属于不同层级。若业务要求可验证的数据清除，需要建立覆盖各层的生命周期协议，并记录完成证据。

实验使用合成商品数据，只验证查询投影中的删除传播和旧事件防倒退。它没有处理用户个人数据删除、日志脱敏或备份擦除，不应从一次查询结果为空推出合规结论。读者可以借助同样的事件身份和世代概念建立追踪，但需要另加真实删除范围与验收方法。

权限撤回不能依赖异步投影的乐观假设

企业检索中，权限变化往往比普通内容修改更敏感。若撤回授权后仍允许旧投影命中，就可能产生短暂越权。设计上可以在读取时查询权威权限，或使用短时缓存与版本栅栏，让旧授权无法单独决定访问。权限数据和内容数据若经不同 topic 传播，还要考虑到达顺序。

这些是设计讨论，本书没有搭建身份系统，也没有权限撤回传播实验。因此我们不声称已验证“撤权立即生效”。商品 active 字段不是访问控制列表，隐藏商品也不是授权协议。若把本书链路用于企业知识库，必须增加真实权限服务、撤回事件、读取拒绝和延迟边界的独立测试。

删除故障怎样定位

先确认源库是否真实删除，再确认连接器是否产生 d 包络，随后检查同键 tombstone 与投影最后 offset，最后检查查询是否固定排除 deleted。不要直接在查询库手工删行作为唯一修复，因为下一次重放可能再次出现。若逻辑错误已经影响大量对象，修正消费者后新建世代重建通常比逐行补丁更容易验证。

本章练习

设计一个删除标记清理策略，明确允许回放的最早位置和旧世代保留时间。然后解释为什么“Kafka 已开启 compaction”不能作为投影不会复活旧对象的充分证据。

一个可空列带来的两种状态

源表最初没有 note，后续真实执行 ALTER TABLE 增加 VARCHAR 可空列。数据库中旧行立即具有该列的默认 null 语义，但没有因此为每一行产生一次普通 UPDATE。连接器会学习新的表结构，后续行事件能够携带 note；已经存在于查询库中的旧文档却仍然可能完全没有这个键。

缺少字段与字段值为 null 在许多查询语言中并不相同。即使业务暂时把它们视为等价，完整对账也需要统一表示。因此本实验把 schema 迁移拆成明确动作：先新增源列，再对 live 投影已有文档显式补入 note:null，然后消费后续包含新字段的更新。这个回填由我们的投影迁移 SQL 完成，不是 Debezium 自动更新全部历史行。

真实演练怎样进行

控制程序执行源库 ALTER，再更新 id 一，把 note 写为可识别字符串并递增 revision。然后对 live 世代执行 JSONB 文档归一化，最后启动消费者读取真实 binlog 更新。检查要求收到更新操作，查询 id 一的 note 等于预期字符串，并将所有源行与投影行完整比较。

随后重建连接器创建新快照，新世代直接读取含 note 的完整表结构。新世代与 live 世代比较相等，证明显式迁移后的在线文档和重新生成的文档具有同样语义。这个组合验收比只检查新增字段能被 JSON 解析更强，也能发现旧行缺少默认键的问题。

为什么完整替换有助于理解

消费者从 after 构造当前完整业务文档，已知基础字段固定读取，note 存在时加入。它没有直接把任意新字段全部暴露给查询层。这样字段变更仍需经过代码评审，避免源表突然增加敏感列就自动进入可搜索副本。代价是每次新增查询字段都需要更新投影契约与兼容逻辑。

对已知字段完整替换，可以避免字段被源端清空后投影仍保留旧值。若采用局部 patch，需要区分“没有提供字段”和“明确设置 null”，还要定义删除字段的行为。不同查询引擎对 null 和缺失的索引策略也不同，替换存储后必须重新验证，不能只比较 Python 字典输出。

兼容窗口比一条 DDL 更长

生产上线通常需要旧消费者和新消费者短暂共存。安全的加字段流程可以先让读取端容忍缺失和 null，再增加源字段，部署写入与投影支持，完成历史回填，最后启用依赖新字段的功能。每一步都应当能够回滚或暂停，不让新功能先于数据准备完成。

删除字段、修改类型和重命名通常比新增可空列更危险。重命名可能在事件上表现为旧字段消失、新字段出现，消费者若仍强制读取旧名会失败。类型变化可能改变 converter 的编码和精度。当前实验只验证新增可空字符串，不声称已经覆盖任意 DDL 自动迁移。

结构历史也是恢复数据

连接器需要正确表结构才能解释日志。清空 schema 历史后尝试从旧 binlog 位置继续，可能让解码缺少上下文。连接器内部历史 topic 应当按官方要求维护，而不是把它当成可以随意压缩和删除的普通业务 topic。我们给在线和重建连接器分别配置历史 topic，避免互相覆盖。

消费者自己的文档版本也值得记录。当前代码规模小，通过发布版本和固定字段契约追踪；更复杂系统可以在投影元数据中保存 schema version，使读取层明确处理旧版本。版本字段不能仅仅用于展示，还应参与迁移判断和验收，避免一批文档永久停留在旧结构。

回填必须有独立验收

手工回填能修复结构差异，也可能覆盖刚刚到达的新值。实验在受控有限阶段先完成源更新、再统一补 null、后消费新消息，因此最终新值不会被回填覆盖。持续生产流中不能直接照搬这个顺序，需要使用只补缺失字段的条件、分批迁移和并发版本保护，或者选择新世代重建。

本书的迁移 SQL只用于当前实验时间线。它证明新增列、显式默认值迁移和后续真实事件能够共同收敛，不是通用在线无锁迁移框架。部署前必须根据实际并发写入和查询行为补充测试。

本章练习

将 note 改为必填字段，给出源库、旧消费者、新消费者、历史文档和读取功能的上线顺序。再设计一种只为缺失键补 null 的迁移条件，解释它为何比无条件覆盖更适合持续消费场景。

为什么恢复不能只有续读

消费位点仍然存在，未必说明当前投影值得继续使用。早期消费者可能遗漏删除，字段转换可能有错误，运维也可能误改查询数据。继续消费只能应用未来变化，无法自动修复所有历史偏差。可靠系统需要一条从权威源重新生成查询状态的路径，使正确性不依赖对旧故障的完整猜测。

重建不应直接清空在线投影。清空后慢慢补数据会让用户看到大量缺失，失败时也没有可回退版本。我们使用 generation 将新旧投影隔离：live 继续保留，新世代 new-generation 接受另一条完整快照。两个世代共享数据库实例，但主键命名空间不同，因此重建失败不会删除在线查询对象。

新连接器、新流与新消费组

实验创建 book-rebuild 连接器，使用不同的复制 server id、topic 前缀和 schema 历史 topic。它对当前源表执行真实 initial 快照，产生 rebuild.inventory.products。新的 rebuild 消费组写入 new-generation，而不是复用 main 组和旧历史。这样新快照不会被旧事件去重状态挡住。

绑定表把每个世代固定到一个消费组和一个 topic。若读者尝试用 main 为新世代回放，程序启动时明确拒绝。这不是限制重建能力，而是要求重建带着新的输入身份发生。Kafka offset 是流内位置，不是能够跨 topic 复用的业务版本。

重建的验收顺序

首先要求新连接器确实产生快照读事件，数量等于当前源表行数。然后读取源库全部字段，与新世代未删除文档逐行比较。最后比较 live 与 new-generation 的查询集合，要求旧世代仍保持完整且两者相等。连接器与任务状态也必须都为 RUNNING，避免刚好读完部分数据后任务已经失败却未被发现。

这里的源写入在重建阶段保持停止，因此完整比较拥有稳定参照。实验验证了独立全量重建、schema 默认值和旧世代保留，没有实现持续写入期间的无缝原子切换。标题中的可重建索引强调数据派生与验证能力，具体查询存储为 PostgreSQL，不涉及外部搜索引擎 alias 的原子性。

持续写入时还需要什么

生产重建通常要在新快照后继续追增量，直到新世代达到明确切换水位。切换动作本身也需要一致性：可以让查询配置指向一个世代，变更时原子替换指针，或者使用查询引擎提供的别名能力。每种实现都要验证正在进行的请求、缓存与回滚路径，而不是只更换一条环境变量。

若旧世代与新世代使用不同转换逻辑，不能要求它们字节完全相同，而应根据新契约进行对账。例如新增字段、删除敏感内容或改变索引表示时，应当把差异写成明确的转换规则。没有规则的差异是故障，有规则的差异仍需真实验收，不能笼统称为“版本升级所以不同”。

回滚与回收是两个时刻

新世代通过对账后，也不宜立即删除旧世代。保留一段观察期可以在查询性能或业务语义出现问题时快速回退，但要限制旧世代的使用资格，避免之后误切回过时数据。回滚若要保持新写入可见，需要旧世代继续追流，或者接受明确的数据延迟边界。

最终回收旧世代之前，应检查查询指针、后台任务、缓存和导出程序都不再引用它，再清理对应消费组与恢复元数据。清理某一张文档表并不自动释放 Kafka 组状态和历史表空间。当前实验只在隔离容器结束时整体清理，没有提供生产世代回收调度器。

重建也要考虑源库压力

重建连接器会再次扫描源表，并与在线连接器共同消耗资源。大型业务不能因为查询副本“可丢弃”就无限重复重建。需要估算扫描时间、额外日志生成、快照事务存活、网络带宽与投影写入成本。失败后从头重建是否可接受，也应当在服务恢复目标中有明确答案。

本书通过两千行和完整比较建立方法，不提供大表重建耗时承诺。生产可分租户、分范围建立世代，或者选择增量快照，但每次拆分都要保留集合覆盖和删除处理的证明。只把全表拆成多个任务，没有共同水位，可能形成一个从未真实存在过的混合视图。

重建完成后还要验证读取路径

数据在新世代中正确，并不代表用户已经读到它。查询配置可能仍然指向旧世代，缓存可能持有旧结果，导出任务也可能绕过统一入口。生产切换验收应使用实际查询接口读取已知新增、更新和删除对象，再核对世代标识。当前实验只比较数据库文档，没有一个生产商品查询服务，因此不会把数据库内的新世代相等称为端到端线上切换已经完成。

本章练习

为持续写入的商品系统设计世代切换协议，写出快照完成、增量追平、验收、指针切换、观察与回收六个边界。再指出本书哪些边界已经有真实实验，哪些仍需要你新增验证。

从行数开始，但不要停在那里

源表和投影都显示两千行，看起来令人放心。实际上源表可能有 id 一到两千，投影却缺少 id 二、多出一个已删除对象，行数仍然相同。再加上字段错位、价格单位和 null 语义，单纯数量比较远远不足。行数适合作为快速探针，不能作为最终正确性证明。

本书从源库按 id 排序读取全部业务字段，从投影读取相同世代里未删除的 document，并按 id 排序。实际通过条件是两个完整列表直接相等。随后才把规范 JSON 序列化后的 SHA256 写入报告，便于跨运行核对。摘要帮助压缩证据，却不替代程序真正执行的内容比较。

比较前先统一语义

整数价格、整数 active 与 revision 在两端保持相同类型。新增 note 后，源库旧行的 null 必须在投影中表现为同样键和值。若一边没有键，另一边键为 null，当前契约会判为不同。这种严格性有助于发现 schema 迁移遗漏，也要求我们先定义什么叫等价。

真实生产可能存在时区、精度、排序、字符归一化和默认值问题。对账程序应共享业务契约，而不是共享消费者的全部实现。如果两者复用同一个错误转换函数，可能一起把错误当正确。实验源侧直接读取数据库行，投影侧直接读取文档，没有用消费事件重新构造源侧预期，从而减少这种同源错误。

六次比较覆盖不同阶段

初始快照加增量比较验证接入衔接；重复回放比较验证幂等；崩溃重启比较验证重复恢复；停机追平比较验证积压；schema 演进比较验证新增字段与默认值迁移；全量重建比较验证新世代。每次都保存单独报告，使失败可以定位到某一阶段，而不是只在最后发现所有结果都不对。

对账也会检查删除后的结果，因为源库不存在的对象不会出现在源列表，而投影查询固定排除 deleted。若消费者没有删除标记，或者查询忘记过滤，列表长度或具体 id 都会不一致。这个检查覆盖业务可见性，但不会检查内部删除文档是否已经从磁盘擦除，那是不同目标。

对账时的时间参照

两个数据库之间没有共同读事务。如果源库在读取过程中继续变化，而投影在另一个时间被读取，结果差异可能来自时间边界，而不是永久错误。实验为有限批次停止写入后进行比较，因此参照稳定。生产持续系统需要固定水位、快照、变更窗口或允许重试的差异分类，不能每次比较都把短暂延迟当成数据损坏。

一种做法是按对象业务版本判断是否已经可比较：只核对源版本不再变化并且投影已经达到相应版本的对象。另一种是建立一致快照与日志截止边界。这些方法需要来源支持，不能简单取两个系统各自的当前时间。时钟同步有帮助，但不能把独立提交变成原子提交。

大规模对账如何扩展

数据量大时可以按稳定主键范围分段，记录每段行数、键集合摘要和字段摘要，先定位差异段，再逐行钻取。分段必须防止范围遗漏和重叠，不能只比较随机样本。哈希摘要还应包含字段名称和明确的编码规则，避免拼接歧义；例如两个字符串字段直接连接会让不同元组产生相同字节串。

本书使用结构化 JSON、固定排序与明确字段，因此可重复生成摘要。SHA256 相同仍是基于哈希抗碰撞假设的压缩证据，实际实验还比较原始列表。生产是否保留逐行明细，要结合敏感数据和证据体积决定，可以保留差异键与脱敏字段，而不是把整库内容上传到公开 CI。

发现差异之后不要立即补丁

先保留源快照、投影状态、最后交付位置与相关消息，再判断差异类型。少量对象落后可能是延迟，永久缺失可能是位点跳过，旧值覆盖可能是键路由或顺序错误。直接在投影上手工修复虽然能让一次比较通过，却可能被之后重放再次覆盖，也可能掩盖系统性错误。

修复应当对应原因：消费者逻辑有误则发布修正并重建，个别消息失败则按保留的输入安全重试，源数据错误则在权威源修复。对账是定位与验收工具，不应变成绕过同步协议的第二写入入口。

差异报告要避免泄露源数据

公开CI适合合成数据，不适合把真实客户行直接上传artifact。生产对账可以记录差异主键、字段名、版本和受控摘要，把敏感原值留在受权限保护的系统里。摘要也可能受到低熵字段枚举推测，不能自动当成脱敏。证据设计需要同时满足定位与最小暴露，两者都应在上线前确定，而不是出故障后把整库导出到临时公共位置。

本章练习

构造两份行数相同但内容不同的数据，说明数量和最大 id 为什么都会通过。再写一个分段对账方案，明确边界、规范编码、差异明细和持续写入期间的重试条件。

让排查从问题对象开始

用户报告 id 一价格不对时，先确认源库当前值和最后业务修订，再沿连接器、Kafka、消费者、投影与查询依次追踪。不要先重启全部组件，因为重启可能改变现场并增加重复，且无法回答哪一层最先偏离事实。一个稳定对象 id 比一个模糊的“同步不正常”告警更容易建立证据链。

源库侧检查真实行、事务提交与 binlog 配置。连接器侧检查任务状态、错误堆栈与源位点。Kafka 侧检查 topic、分区、键与目标记录。消费者侧检查组位点、历史表、最后投影 offset 和错误日志。查询侧检查世代选择与删除过滤。找到第一处缺失或错误的边界，才能选择合适恢复动作。

RUNNING 是必要条件，不是充分条件

Connect 的 connector 处于 RUNNING，不代表每个 task 都正常，也不代表业务数据已经完整到达。本实验最终同时检查连接器和全部任务的状态，并结合消息操作种类和完整对账。若任务失败但 connector 外层仍存在，只看一个顶层状态就会误判。

同样，Kafka 消费者进程活着可能只是持续轮询空 topic，也可能订阅了拼写错误的前缀。早期实验真实遇到过 topic 尚未创建时订阅失败，因此控制程序明确等待连接器数据 topic 出现。等待有时间上限，缺失会失败，不会把不存在的 topic 自动变成“无数据所以成功”。

这份固定镜像已观察到的启动提示

基线容器日志出现ConnectRestExtension插件发现错误，具体是io.debezium.connector.mysql.rest等MySQL、PostgreSQL、SQL Server和MongoDB扩展Provider未找到。该次固定Connect镜像的rest.extension.classes为空，实验只使用标准Connect管理API与MySQL源连接器；标准API、真实binlog数据路径和二十二项检查均实际通过。我们保留这些原始日志，没有把所有ERROR行过滤掉。

这个结论只限定于已观察的可选REST扩展发现提示，不能推广为任意ServiceConfigurationError或堆栈都可忽略。若标准API不可用、MySQL源插件未装载、任务不为RUNNING或消息缺失，实验仍须失败。升级镜像时应重新检查该提示是否变化，并继续验证实际数据路径，而不是根据旧报告静默豁免新错误。

从错误形状识别配置问题

真实 JSON 消息如果出现 schema 与 payload 包装，而消费者期待顶层 op，会报字段缺失。这通常提示 converter 配置或解析契约不一致，不能通过忽略所有缺少 op 的消息掩盖。应当保存收到的结构，确认有效配置，再更新配置或版本化解析器。

如果出现数据库唯一冲突，需要先判断它是预期去重还是意外键冲突。delivery_history 的 ON CONFLICT 是设计内路径；consumer_binding 的冲突则需要读取现有绑定并拒绝不匹配参数。把所有完整性错误统一吞掉，会让错误世代看起来消费完成，实际却没有任何文档。

把日志写成可关联的信息

有效日志至少带对象身份、topic、partition、offset、消费组与世代。涉及源位置时还要保留 binlog 文件与位置的命名空间。不要只打印“update failed”，也不要无差别输出包含密码的完整连接配置。教学证据中的固定本地密码没有生产价值，实际环境需要脱敏和秘密管理。

本实验消费者报告消息总数、新鲜数、重复数、操作计数和少量 source 样例，控制程序另存检查列表与容器日志。消息数量帮助发现路径是否发生，逐行比较负责最终内容。日志与报告互相补足，不能互相替代。

失败分类决定恢复方法

依赖启动失败通常先修环境；连接器权限不足先核对最小权限与快照要求；schema 不兼容先停在明确失败点并发布兼容修正；投影暂时不可用可以在未提交位点条件下重试；历史位点超出保留范围则需要重建决策。把这些情况都写成无限重启，会持续消耗资源却无法恢复缺失事实。

对于毒消息，生产可以建立有审计的隔离流程，但不能在没有业务决策时自动跳过。跳过意味着查询副本可能永远缺少该对象的某次状态；后续完整文档更新有时会修复，有时不会，尤其删除缺失可能长期残留。本书采用遇到不支持事件就失败的策略，没有实现死信队列，因此不会声称验证了隔离后继续处理的正确性。

一次故障的结束条件

修复后组件变绿只是开始。需要确认 backlog 下降、关键探针可见、删除路径正常，并执行对账或有边界的业务验收。若通过新世代重建恢复，还要确认查询实际指向新世代，旧版本没有被后台任务重新启用。最后保存产品版本、输入位置、恢复动作与验收结果，供下一次故障比较。

实验报告关联 GitHub run URL，使检查结果能够回到具体源码提交和运行环境。公开报告只使用合成数据，因此可以完整分享；生产报告需要限制敏感字段，但仍应保留足够的事件身份来重现判断过程。

本章练习

为“源库已删除，查询仍返回对象”写一份最短排查路径。每一步列出读取证据、可能结果和下一步，禁止用“重启试试”代替判断。再说明在哪一步可以区分连接器没发删除与查询忘记过滤。

先计算哪些数据会增长

源库表会增长，binlog 会增长，Kafka 数据与内部 topic 会增长，PostgreSQL 投影、去重历史和索引也会增长。把投影文档大小乘以行数，只覆盖其中一部分。特别是 delivery_history 按事件增长，一个频繁变化的小表也可能形成很大的历史表；删除商品不会自动删除它之前所有交付坐标。

因此容量模型至少区分当前对象数、每日变化数、消息平均大小、峰值、保留天数与副本数量。Kafka 压缩和数据库索引还会改变实际空间，不能仅用 JSON 字节数估算磁盘。实验保存有限合成数据，不提供通用压缩比或每核吞吐数字；生产应当用自己的字段分布和更新模式测量。

延迟由多段组成

源事务提交后，连接器需要读取并转换，broker 需要接受，消费者需要轮询和写数据库，查询系统还可能有缓存。平均端到端延迟很低，可能掩盖某一段偶尔停顿数十秒。监控应当既能看整体业务探针，又能按阶段拆分，避免一遇到延迟就盲目扩容 broker。

连接器批量参数、Kafka 批处理、消费者事务大小都会影响延迟与吞吐。更大批次可能减少往返，却增加等待和故障重放量；每条同步提交清晰但开销较高。任何调整都应当保留提交顺序和幂等不变量，再用真实负载比较，不能因为某个参数名字带 throughput 就直接调到最大。

历史表的索引成本

delivery_history 的复合主键保证事件身份唯一，但每条事件都要进行索引写入。投影 upsert 也需要访问商品主键和旧版本行。高更新率下，数据库清理、事务日志与磁盘延迟可能先于 CPU 成为瓶颈。只看消费者进程 CPU 很低，不能推断它没有性能限制。

可以考虑按时间或输入范围分区历史表，并在明确恢复边界之后回收旧分区。然而当前主键是事件坐标，不含处理时间，增加生命周期字段需要同时设计约束与清理验证。清理之后能否安全接受旧回放，要看投影是否仍保留足够水位以及业务操作是否纯覆盖。本书没有实现这项扩展。

分区扩展带来的协议成本

单分区限制并行度，多分区增加吞吐空间，但同对象顺序仍需要稳定键路由。数据倾斜时一个热门对象无法通过增加分区自动拆散，除非业务允许分解状态。分区变化还会影响本书 offset 条件，不能简单修改 topic 分区数后继续使用原投影。

一种可审计的升级方式是新建目标分区数的 topic，通过新连接器或明确迁移管线建立新世代，再对账和切换。这样把路由变化变成一次显式版本迁移，而不是悄悄改变现有位置的含义。代价是重建时间和双份资源，需要提前纳入容量预算。

快照与追赶争用资源

全量快照可能占用源库缓存与磁盘，在线 binlog 读取也需要及时前进。若快照速度很慢，日志留存不足，重建尚未完成就可能失去衔接需要的日志。只增加 Kafka 磁盘无法解决源库日志清理问题，因为变化还没有进入 Kafka。容量规划必须沿整条链路计算最薄弱的恢复窗口。

投影重建也可能与线上查询争用同一 PostgreSQL 实例。generation 隔离的是逻辑身份，不是 CPU、IO 或连接数。生产若对可用性要求更高，可以使用独立重建集群或资源限额，但需要验证切换方式。我们不把同实例不同世代描述成资源故障隔离。

如何建立有意义的压测

先保留本书的正确性检查，再增加数据量和写入模式。负载应当包含插入、更新、删除、热点对象、大文档与 schema 兼容路径；只连续插入短字符串会高估许多系统的表现。每次测试记录环境、参数、持续时间和是否达到稳态，并在结束后完整或分段对账。

测试中还应安排消费者停机，观察恢复净吞吐是否足以追平；安排投影慢写，观察连接器与 broker 的缓冲边界；检查异常退出后是否仍然收敛。性能提升若依赖提前提交位点或丢弃失败消息，应当判为正确性失败，而不是一种可接受优化。

资源隔离需要真实限制

逻辑上分开世代并不会限制重建连接数，也不会防止一个错误循环耗尽磁盘。生产运行器应有CPU、内存、连接、队列和磁盘预算，并在达到阈值时停在可恢复位置。单纯把超时设得很大，只会延后暴露资源耗尽。增加这些限制时也要测试失败路径，确认不会在投影尚未提交时推进位点；资源保护只有与正确性协议一起工作，才真正提高可恢复性。

本章练习

按每天一千万条变更、每条一千字节、保留三天估算原始日志量，再列出未包含的副本、索引、压缩和峰值因素。为当前单分区消费者设计一项批处理优化，写明必须保留的事务与位点顺序。

上线不是把 Compose 搬到生产

教学环境使用固定本地密码、单节点 broker 和明文内网连接，目的是减少理解成本。生产需要依据实际组织边界配置身份、加密、网络限制和秘密管理，还要规划高可用与备份。这里讨论上线设计，不声称随书实验已经验证这些生产设施。复制一份配置文件无法替代容量、权限和故障演练。

接入源库之前先检查 ROW 日志、行镜像、复制身份、日志保留和快照权限。对已有数据库修改配置可能影响其他复制任务，应由数据库维护流程评估。连接器 server id、topic 前缀、内部 topic 和消费组必须有命名与所有权约定，避免不同应用在同一个集群中发生冲突。

把输入、转换和读取分开发布

连接器版本、消费者转换版本与查询功能版本可以分别变化。若把它们锁成一个不可观察的大开关，出现错误时很难判断来源。更稳妥的方式是先让读取端兼容新文档，再部署新转换，验证数据准备，最后启用业务功能。每个版本都应记录使用的输入契约和支持的 schema 范围。

对于不兼容转换，优先建立新世代而不是原地覆盖全部在线对象。新世代可以接受完整对账和业务查询验收，旧世代仍可保留以便回退。当前实验已经验证隔离重建和内容相等，但没有提供生产查询指针服务，因此真正上线时需要实现并测试切换原子性。

位点操作需要成为受控动作

消费组 reset、连接器重置与内部 topic 清理都会改变恢复依据。它们不应隐藏在“重新部署”脚本里。执行前保存原位置、目标位置、原因、预计影响和回退方案，执行后通过业务探针与对账确认。特别是向前跳过位置，可能造成永久缺失，不能仅以 lag 变小作为成功标准。

本书用 –replay 从起点消费属于受控实验，且同组历史完整保留。生产回放应使用专门恢复流程，避免与正常组成员并发竞争。新世代通常使用新消费组更清楚，也能让绑定关系拒绝错误复用。所谓“复用方便”如果让输入身份失去含义，后续恢复成本会更高。

交付物必须能对应同一个源码

仓库保存书稿、实验源码、依赖和构建脚本。流水线从同一提交运行真实实验，生成 EPUB 和分章网页，验证内容一致、下载 ZIP 和移动端，再部署 Pages。Release 使用这次成功 CI 已经生成的下载产物，不能在本地重建另一份同名 EPUB 后上传，否则版本号相同也可能字节不同。

我们记录 EPUB 字节数与 SHA256，并比较 CI artifact、Pages 公共下载和 Release 公共资产。EPUBCheck 对同一产物执行验证。源码提交说明实验由什么代码生成，运行证据说明发生了什么，字节摘要说明读者拿到什么。三者彼此关联，才能避免“源码正确但下载仍旧版”的常见交付问题。

文档更新与产品构建分开

交付说明和 canonical 证据回写使用跳过 CI 的文档提交，工作流也按产品输入路径过滤。否则单纯更新发布日期可能触发重建，造成 Pages 上的 EPUB 与已有 Release 不同。产品输入发生变化则应重新验证，并按版本规则发布新补丁，保留旧资产，不覆盖同一已发布版本。

随书 ZIP 排除 evidence 运行目录和缓存，防止旧的 kill 标记或临时日志进入下一次复现。证据通过独立报告和 artifact 分发，实验包只包含重现所需源码、SQL、Compose、依赖和说明。独立解压后完整重跑，比仅检查文件存在更能发现路径和依赖遗漏。

上线验收要面向业务

技术验收包括真实链路、重复恢复、删除、schema 和重建；业务验收还包括筛选结果、排序、可见性、字段单位与异常提示。PostgreSQL 文档相等不能证明某个前端展示没有把分当元，也不能证明权限过滤正确。本书的在线阅读界面验收与数据投影验收属于不同对象，不能互相充当证据。

生产发布后保留观察窗口，检查错误率、延迟、源库负载和对账差异，再回收旧资源。若发现问题，优先暂停新功能或回到已验证世代，同时保留输入日志和现场。不要为了让发布单快速关闭而删除全部旧状态。

本章练习

写一份新字段上线的发布单，包含输入配置、转换版本、历史迁移、查询兼容、对账和回退。再说明为什么一个仅修改 DELIVERY 的提交不应该自动生成新 EPUB。

用一条完整时间线串起全书

实验首先创建源表和两千行合成商品，初始化空投影。连接器开始真实 initial 快照，控制程序观察实际快照 SELECT，在同一查询仍活跃期间完成插入、更新、删除。随后消费者读取 r、c、u、d 事件，把数据写入 live 世代，执行第一次完整对账。这个起点同时覆盖已有数据与并发增量。

接着同组从起点显式重放，要求全部记录被识别为重复，结果保持相等。三种错误绑定必须在处理前被拒绝。新的 crash 组在数据库提交后、Kafka 位点提交前进入屏障，控制程序检查两个存储的真实状态，再发送 SIGKILL。普通重启至少遇到一条重复，并最终得到正确 crash-gen 投影。

从停机恢复到结构迁移

下一阶段没有任何 live 消费者运行，源库执行一百次更新、一次删除和一次插入。控制程序先证明投影已经落后，再启动 main 组追赶，验证真实操作计数和完整内容。这个顺序避免“没有产生故障也通过”的空测试，让恢复前后的差异可见。

随后真实新增 note 列，更新一行，显式迁移 live 文档的 null 默认值，再消费新 binlog。对账要求全部行字段一致。最后创建独立重建连接器、新 topic、新消费组和新世代，从当前源库重新快照，验证重建结果与源表相等，同时旧 live 世代仍保留且内容一致。

每一项检查应当支持具体结论

快照查询前后观测支持写入交叠，操作计数支持实际经过四条事件路径，binlog 样例支持真实来源，事务历史与 broker 位点支持故障窗口，退出码支持真实强制终止，重复计数支持重投，完整对账支持最终集合相等，绑定拒绝支持参数不变量。任何一项都不能独自替代其他项。

例如最终对账相等，可能只是错误测试根本没执行更新；因此要检查操作路径。进程被杀，也可能还没有处理消息；因此要检查数据库提交和 broker 未确认。连接器运行，也可能订阅错表；因此要检查真实消息和源投影内容。把检查拆成这些独立证据，才能评估测试到底覆盖了什么。

初次运行的实际基线

本书早期完整基线在 Linux GitHub Actions 上运行，真实 MySQL 为八点四点十一，Debezium 为三点一点二 Final，PostgreSQL 为十七点十一。基线通过二十二项检查，集成本体约一百零三秒。最终出版运行还会重新执行并记录自己的时间与环境；这些数字用于说明实验确实发生，不是生产性能承诺。

镜像标签后来固定为该次运行取得的真实摘要，Kafka 版本还由容器内命令输出保存。Python 依赖固定版本。读者复现时应检查运行证据与下载包是否对应同一发布版本，不要把另一个分支上的报告当成当前源码已经通过的证明。最终 canonical 报告以下载区 validation 为准。

仍然没有被验证的故障

本书没有让 broker 多节点选主，没有制造 MySQL binlog 已被清理后的缺口，没有数据库断电，也没有跨机房分区。我们没有测试多分区重平衡、跨表事务可见性、主键迁移、任意 DDL、权限撤回和搜索引擎排名。真实单机实验能建立边界清晰的基础，不能凭绿色 CI 延伸成这些更大的保证。

这些限制不是让读者停止使用，而是帮助下一轮实验有明确目标。若部署环境最担心日志丢失，就优先演练连接器恢复失败与重建；若最担心跨租户泄露，就增加真实权限服务与查询拒绝；若最担心吞吐，就在保留正确性检查的前提下增加数据量和故障期间负载。

为团队建立周期演练

周期演练可以按风险挑选场景，每次保留源码版本、环境、故障触发点、观察证据和验收结果。避免只在上线前执行一次，之后依赖版本和配置不断变化却不再复查。镜像升级、连接器升级、字段变化、分区策略变化和历史清理都可能改变恢复假设。

演练结束时应回答三个问题：哪些数据已经证明正确，哪些路径仍未验证，下一次故障由谁根据什么证据决定回放或重建。一个可操作的恢复手册，比一页笼统的“支持高可靠同步”说明更有价值。系统的可信度来自这些可以重新产生的事实。

把恢复手册交给没有参与开发的人

一次有价值的交付评审，可以让另一位维护者只凭仓库说明与下载包重新运行。他应当能够找到入口、理解资源所有权、读出失败原因，并判断报告是否对应当前源码。如果必须依赖作者记忆中的隐藏命令或本机目录，恢复路径就还没有真正产品化。独立解压执行是其中一项检查，无法替代全部运维演练，但能及时暴露最常见的依赖遗漏。

同样，书稿中的每个保证都应当能指向代码和证据。遇到无法定位的句子，应当补实验、缩小结论或明确标记为设计讨论。技术内容的深度不来自承诺数量，而来自边界是否解释清楚，以及读者能否在自己的条件下重新检验。持续维护这样的对应关系，才让可重建从一句架构口号变成实际工程能力。

本章练习

选择一个本书未验证但与你的业务直接相关的故障，写出真实触发方式、预期损坏、需要保留的证据和通过条件。禁止只增加一个 mock 测试就声称完成集成覆盖；先说明测试与生产边界之间还有哪些差距。

第一至三章：契约与环境

用户目录应以稳定用户身份为主键，明确显示名、状态与组织字段的来源；禁用通常保留事实但改变可访问性，物理删除则需要处理所有副本。权限敏感场景应由权威授权决定读取，不能只依靠异步目录。五个提交边界分别是源库、连接器交付、broker、投影、消费组；Kafka有记录而投影没有，可能是消费者未读取、事务失败或错误世代，分别检查组位点、错误日志与绑定。

版本查询证明实际服务器版本，镜像摘要证明运行内容身份，插件列表证明Connect装载的组件版本。它们不能证明安全性或生产兼容性。端口冲突实验应在专属环境进行，启动失败后只清理该Compose项目，不应终止占用端口的未知服务。

第四至六章：身份、快照与水位

多租户应在源主键、Kafka键与投影主键中共同体现身份，或者明确全局唯一id保证。仅在文档里加tenant_id不能阻止同id覆盖。移除SLEEP后交叠检查可能来不及观测快照，这是测试同步条件不满足，不是连接器算法错误。替代方案需要可观察的真实快照阶段和写入确认，不能只增加固定睡眠。

不同分区的offset独立增长，不能取最大值当作全局水位。增加分区前需要验证键映射、旧记录顺序、投影比较条件与切换；安全路线可以新建topic和世代，而不是原地改变位置的意义。

第七至九章：包络、幂等与崩溃

删除包络写入删除标记，tombstone只推进交付元数据，普通更新替换已知文档。converter包装变化应明确升级契约或解析器，不能忽略所有缺少op的记录。积分加一需要事件级去重与累计更新共享事务，否则重复会重复计数。

当前事务消除了历史已提交而投影未提交的窗口，保留投影成功而位点未确认的重复窗口。把屏障放到Kafka commit之后，重启可能不再重复，但最终仍然相等；因此最终结果相等并不能独自证明测试经过了位点前故障。

第十至十二章：积压、删除与结构变化

示例积压三十六万条，恢复净速度为每秒二百条，理想追平约一千八百秒。数据库锁、网络延迟、批处理、热点与分区限制都会延长时间；增加实例不能拆分单分区。删除标记清理必须约束最早回放位置和旧世代寿命，Kafka压缩不能阻止查询库单独复活旧对象。

必填字段迁移应先部署兼容读取与写入，再补源历史值、支持投影和回填，验收完成后启用强约束功能。只补缺失键的迁移条件应判断文档中不存在note，而不是把已有值无条件覆盖；持续消费还需并发保护和回归验证。

第十三至十五章：重建、对账与定位

持续重建需要独立快照、增量追平、明确切换水位、实际查询验收、原子指针与回收协议。本书真实验证了隔离快照、完整内容与旧世代保留，生产无缝切换仍需实现。数量相同和最大id相同都无法排除中间缺行或字段错误，分段对账需要无遗漏范围和规范编码。

删除排查先查源事实，再查连接器d事件、Kafka同键记录、投影最后offset与deleted，最后查查询过滤。看到Kafka已有删除但投影没更新，可继续查消费者；投影deleted正确但接口仍返回，则更接近读取路径问题。

第十六至十八章：容量、发布与演练

每天一千万条、每条一千字节约十GB原始数据，三天约三十GB，尚未包含副本、索引、内部topic和空间余量。批处理必须保证整批数据库事务完成后才提交末尾位点。交付文档不属于产品输入，更新它不应悄悄改变已发布EPUB。

新增故障演练应使用实际组件，例如主动清理所需binlog后检查连接器失败，再执行真实重建与对账。mock可以帮助验证局部分支，却不能替代复制日志已经丢失时的集成恢复证明。每项演练都应写清故障触发点、保留证据和适用边界。

lab/README.md

# 真实 CDC 实验

需要 Linux/macOS Docker Compose、Python3.12、至少约4GB可用内存、可访问镜像与Python依赖源。端口23316/29092/28083/25436必须空闲。凭据仅限本地教学；不要在已有业务数据库上执行初始化SQL。

```sh
export CDC_COMPOSE_PROJECT=cdc-book-learning-001
sh lab/run.sh
```

入口安装固定依赖、创建项目独占容器、保存实际镜像摘要与版本，执行真实MySQL→Debezium→Kafka→PostgreSQL集成，最后只清理指定Compose项目及其卷。不要并行运行两份（端口固定）。失败时evidence保留检查与容器日志，不会skip缺失依赖。重跑应使用空的evidence目录和新的项目身份，避免旧kill-ready.json污染控制信号；正式CI在独立目录执行。

实验依次验证快照期间真实写入、r/c/u/d、重复重放、三种绑定误用拒绝、DB提交而broker未提交时实际SIGKILL/重启、停机100次更新追平、删除、真实ALTER与显式投影默认值迁移、新connector全量重建、六次逐行逐字段对账。runtime.json包含22项检查及实际环境。数据是2000行合成商品；固定镜像digest与依赖见compose.yml和requirements.txt。

消费者是有限实验运行器，读到数据后空闲3秒结束，再由对账判定收敛；不是生产持续服务。单broker/单分区，未验证多机故障、日志过期、跨表原子可见性、任意DDL、权限撤回。PostgreSQL投影不是全文搜索引擎。数据库事务与Kafka位点分开提交，保证为至少一次交付下的幂等收敛，不是端到端exactly-once。

ZIP只包含源码、SQL、依赖、Compose和说明，不携带运行evidence/marker。出版CI会在独立解压目录从空容器环境重跑同一入口。独立证据通过发布artifact和下载区validation.json提供。

lab/compose.yml

services:
  mysql:
    image: mysql@sha256:0f51e63e537a1f176423fa03ef559aabe6fa4372a53160a44bc7e75da7d15364
    command: [--server-id=184054, --log-bin=mysql-bin, --binlog-format=ROW, --binlog-row-image=FULL, --binlog-expire-logs-seconds=86400]
    environment:
      MYSQL_ROOT_PASSWORD: cdc-book-local
      MYSQL_ROOT_HOST: '%'
    ports: ['127.0.0.1:23316:3306']
  kafka:
    image: apache/kafka@sha256:4ceccc577f03f51f6af8dbfda55194d0d892f4fa7913ffbded567ce3895622ed
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,HOST://:29092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,HOST://127.0.0.1:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,HOST:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    ports: ['127.0.0.1:29092:29092']
  connect:
    image: quay.io/debezium/connect@sha256:1290a7abb0c1fe957f8d732e43efbea24201d9ba0fa5ba79582f76438c05d0ba
    depends_on: [mysql, kafka]
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: cdc-book-connect
      CONFIG_STORAGE_TOPIC: book_connect_configs
      OFFSET_STORAGE_TOPIC: book_connect_offsets
      STATUS_STORAGE_TOPIC: book_connect_status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      KEY_CONVERTER_SCHEMAS_ENABLE: 'false'
      VALUE_CONVERTER_SCHEMAS_ENABLE: 'false'
      REST_ADVERTISED_HOST_NAME: connect
    ports: ['127.0.0.1:28083:8083']
  postgres:
    image: postgres@sha256:a63c1ea3f6a3a5a1621f97d2aba3b30faecfab4224c1b0000cc85ced57bb1939
    environment:
      POSTGRES_USER: book
      POSTGRES_PASSWORD: cdc-book-local
      POSTGRES_DB: book
    ports: ['127.0.0.1:25436:5432']

lab/consumer.py

#!/usr/bin/env python3
"""Real Kafka consumer; database transaction commits before broker offset commit."""
import argparse
import json
import os
import signal
import time
from pathlib import Path

import psycopg
from confluent_kafka import Consumer, KafkaError

DSN = 'postgresql://book:cdc-book-local@127.0.0.1:25436/book'


def apply(db, group, generation, message):
    value = json.loads(message.value()) if message.value() is not None else None
    key = json.loads(message.key()) if message.key() else None
    # Kafka tombstones are compaction markers; the preceding Debezium delete
    # envelope, not the null value, carries the source position and operation.
    with db.transaction():
        fresh = db.execute('INSERT INTO delivery_history VALUES (%s,%s,%s,%s) ON CONFLICT DO NOTHING RETURNING 1',
                           (group, message.topic(), message.partition(), message.offset())).fetchone()
        if fresh and value:
            assert value['op'] in ('r', 'c', 'u', 'd'), value
            after = value['after']
            row = after if after is not None else value['before']
            assert key['id'] == row['id']
            document = {k: row[k] for k in ('id', 'sku', 'price_cents', 'active', 'revision')}
            if 'note' in row:
                document['note'] = row['note']
            # Same primary key is always on one partition; no cross-partition
            # numeric offset comparison is claimed here.
            db.execute('''INSERT INTO products_projection VALUES (%s,%s,%s,%s,%s,%s)
              ON CONFLICT(generation,id) DO UPDATE SET document=EXCLUDED.document,
              deleted=EXCLUDED.deleted,source_partition=EXCLUDED.source_partition,
              source_offset=EXCLUDED.source_offset
              WHERE products_projection.source_partition=EXCLUDED.source_partition
              AND products_projection.source_offset<EXCLUDED.source_offset''',
              (generation, row['id'], psycopg.types.json.Jsonb(document), value['op'] == 'd', message.partition(), message.offset()))
        db.execute('''INSERT INTO delivery_checkpoint VALUES (%s,%s,%s,%s)
            ON CONFLICT(consumer_group,topic,partition) DO UPDATE
            SET next_offset=GREATEST(delivery_checkpoint.next_offset,EXCLUDED.next_offset)''',
            (group, message.topic(), message.partition(), message.offset()+1))
    return bool(fresh), value


def bind(db, generation, group, topic):
    with db.transaction():
        db.execute('INSERT INTO consumer_binding VALUES (%s,%s,%s) ON CONFLICT DO NOTHING', (generation,group,topic))
        actual = db.execute('SELECT consumer_group,topic FROM consumer_binding WHERE generation=%s', (generation,)).fetchone()
        if actual != (group,topic):
            raise ValueError('Generation must bind to one unique consumer group and topic')


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--group', required=True)
    parser.add_argument('--generation', required=True)
    parser.add_argument('--topic', required=True)
    parser.add_argument('--crash-after-db', action='store_true')
    parser.add_argument('--ready-file')
    parser.add_argument('--replay', action='store_true')
    parser.add_argument('--idle-seconds', type=float, default=3)
    args = parser.parse_args()
    db = psycopg.connect(DSN, autocommit=True)
    bind(db, args.generation, args.group, args.topic)
    consumer = Consumer({'bootstrap.servers': '127.0.0.1:29092', 'group.id': args.group,
                         'enable.auto.commit': False, 'auto.offset.reset': 'earliest',
                         'enable.auto.offset.store': False})
    def assigned(c, partitions):
        if args.replay:
            for partition in partitions:
                partition.offset = 0
        c.assign(partitions)
    consumer.subscribe([args.topic], on_assign=assigned)
    totals = {'messages': 0, 'fresh': 0, 'duplicates': 0, 'ops': {}, 'source_positions': []}
    last = time.monotonic()
    try:
        while True:
            msg = consumer.poll(0.5)
            if msg is None:
                if totals['messages'] and time.monotonic()-last > args.idle_seconds:
                    break
                if time.monotonic()-last > 60:
                    raise TimeoutError('No messages available; integration may not be skipped')
                continue
            if msg.error():
                raise RuntimeError(str(msg.error()))
            fresh, value = apply(db, args.group, args.generation, msg)
            totals['messages'] += 1
            totals['fresh' if fresh else 'duplicates'] += 1
            if value:
                totals['ops'][value['op']] = totals['ops'].get(value['op'], 0)+1
                if len(totals['source_positions']) < 8:
                    totals['source_positions'].append(value['source'])
            if args.crash_after_db:
                if args.ready_file:
                    Path(args.ready_file).write_text(json.dumps({'topic': msg.topic(), 'partition': msg.partition(), 'offset': msg.offset()}))
                print('database committed; waiting for controller SIGKILL', flush=True)
                while True:
                    time.sleep(1)
            consumer.commit(message=msg, asynchronous=False)
            last = time.monotonic()
    finally:
        consumer.close()
        db.close()
    print(json.dumps(totals, ensure_ascii=False), flush=True)


if __name__ == '__main__':
    main()

lab/projection.sql

CREATE TABLE products_projection (
  generation TEXT NOT NULL,
  id BIGINT NOT NULL,
  document JSONB,
  deleted BOOLEAN NOT NULL,
  source_partition INTEGER NOT NULL,
  source_offset BIGINT NOT NULL,
  PRIMARY KEY(generation,id)
);
CREATE TABLE delivery_checkpoint (
  consumer_group TEXT NOT NULL,
  topic TEXT NOT NULL,
  partition INTEGER NOT NULL,
  next_offset BIGINT NOT NULL,
  PRIMARY KEY(consumer_group,topic,partition)
);
CREATE TABLE delivery_history (
  consumer_group TEXT NOT NULL,
  topic TEXT NOT NULL,
  partition INTEGER NOT NULL,
  offset_value BIGINT NOT NULL,
  PRIMARY KEY(consumer_group,topic,partition,offset_value)
);
CREATE TABLE consumer_binding (
  generation TEXT PRIMARY KEY,
  consumer_group TEXT NOT NULL UNIQUE,
  topic TEXT NOT NULL
);

lab/requirements.txt

PyMySQL[rsa]==1.1.1
confluent-kafka==2.10.0
psycopg[binary]==3.2.9

lab/run.py

#!/usr/bin/env python3
"""Fail-closed integration against real Docker MySQL/Debezium/Kafka/Postgres."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import signal
import subprocess
import sys
import time
import urllib.request

import pymysql
import psycopg
from confluent_kafka import Consumer, TopicPartition
from confluent_kafka.admin import AdminClient

ROOT = Path(__file__).resolve().parents[1]
LAB = ROOT/'lab'
EVIDENCE = Path(os.environ.get('CDC_EVIDENCE_DIR', str(LAB/'evidence')))
EVIDENCE.mkdir(parents=True, exist_ok=True)
checks = []


def check(name, condition, details=None):
    record = {'name': name, 'passed': bool(condition), 'details': details}
    checks.append(record)
    (EVIDENCE/'checks.json').write_text(json.dumps(checks, ensure_ascii=False, indent=2))
    if not condition:
        raise AssertionError(record)
    print('PASS', name, flush=True)


def eventually(fn, seconds=180):
    deadline = time.monotonic()+seconds
    last = None
    while time.monotonic() < deadline:
        try:
            result = fn()
            if result:
                return result
        except Exception as exc:
            last = str(exc)
        time.sleep(1)
    raise TimeoutError(f'Integration dependency did not become ready: {last}')


def mysql():
    return pymysql.connect(host='127.0.0.1', port=23316, user='root', password='cdc-book-local', autocommit=True,
                           cursorclass=pymysql.cursors.DictCursor)


def sql(connection, query, args=None):
    with connection.cursor() as cursor:
        cursor.execute(query, args)
        return cursor.fetchall()


def request(path, payload=None):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request('http://127.0.0.1:28083'+path, data=data,
                                 headers={'Content-Type': 'application/json'})
    with urllib.request.urlopen(req, timeout=15) as response:
        return json.load(response)


def connector(name, prefix, slow=False):
    cfg = {'connector.class': 'io.debezium.connector.mysql.MySqlConnector', 'tasks.max': '1',
      'key.converter': 'org.apache.kafka.connect.json.JsonConverter', 'key.converter.schemas.enable': 'false',
      'value.converter': 'org.apache.kafka.connect.json.JsonConverter', 'value.converter.schemas.enable': 'false',
      'database.hostname': 'mysql', 'database.port': '3306', 'database.user': 'cdc',
      'database.password': 'cdc-book-local', 'database.server.id': '184055' if slow else '184056',
      'topic.prefix': prefix, 'database.include.list': 'inventory', 'table.include.list': 'inventory.products',
      'schema.history.internal.kafka.bootstrap.servers': 'kafka:9092',
      'schema.history.internal.kafka.topic': prefix+'.schema-history', 'snapshot.mode': 'initial',
      'include.schema.changes': 'false', 'topic.creation.default.replication.factor': '1',
      'topic.creation.default.partitions': '1', 'heartbeat.interval.ms': '1000'}
    if slow:
        cfg['snapshot.select.statement.overrides'] = 'inventory.products'
        cfg['snapshot.select.statement.overrides.inventory.products'] = 'SELECT * FROM inventory.products WHERE SLEEP(0.003)=0'
    request('/connectors', {'name': name, 'config': cfg})
    return cfg


def worker(group, generation, topic, extra=()):
    admin = AdminClient({'bootstrap.servers':'127.0.0.1:29092'})
    eventually(lambda: topic in admin.list_topics(timeout=10).topics)
    command = [sys.executable, str(LAB/'consumer.py'), '--group', group, '--generation', generation,
               '--topic', topic, *extra]
    result = subprocess.run(command, capture_output=True, text=True, timeout=150)
    (EVIDENCE/(group+('-replay' if '--replay' in extra else '')+'.log')).write_text(result.stdout+'\n'+result.stderr)
    if result.returncode:
        raise RuntimeError(f'Consumer failed ({group}): {result.stdout}\n{result.stderr}')
    return json.loads(result.stdout.strip().splitlines()[-1])


def source_rows(source):
    rows = sql(source, 'SELECT * FROM inventory.products ORDER BY id')
    return rows


def projection_rows(db, generation):
    return [row[0] for row in db.execute('SELECT document FROM products_projection WHERE generation=%s AND NOT deleted ORDER BY id', (generation,)).fetchall()]


def reconcile(source, db, generation, label):
    a = source_rows(source)
    b = projection_rows(db, generation)
    dump = lambda rows: json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode()
    evidence = {'source_rows': len(a), 'projection_rows': len(b), 'source_sha256': hashlib.sha256(dump(a)).hexdigest(),
                'projection_sha256': hashlib.sha256(dump(b)).hexdigest(), 'all_rows_all_fields_equal': a == b}
    (EVIDENCE/(label+'-reconciliation.json')).write_text(json.dumps(evidence, indent=2))
    check(label+' exact full-row reconciliation', a == b, evidence)
    return evidence


def main():
    start = time.time()
    source = eventually(mysql)
    pg = eventually(lambda: psycopg.connect('postgresql://book:cdc-book-local@127.0.0.1:25436/book', autocommit=True))
    eventually(lambda: request('/connector-plugins'))
    versions = {'mysql': sql(source, 'SELECT VERSION() AS version')[0]['version'],
                'postgres': pg.execute('SELECT version()').fetchone()[0],
                'connect_plugins': request('/connector-plugins'), 'python': sys.version,
                'kafka': (EVIDENCE/'kafka-version.txt').read_text().strip(),
                'images': {name:json.loads((EVIDENCE/('image-'+name+'.json')).read_text()) for name in ('mysql','kafka','connect','postgres')}}
    (EVIDENCE/'versions.json').write_text(json.dumps(versions, indent=2))
    for statement in (LAB/'schema.sql').read_text().split(';'):
        if statement.strip():
            sql(source, statement)
    with source.cursor() as cursor:
        cursor.executemany('INSERT INTO inventory.products VALUES (%s,%s,%s,%s,%s)',
                           [(i, 'sku-'+str(i), i*17, 1, 1) for i in range(1,2001)])
    pg.execute((LAB/'projection.sql').read_text())
    cfg = connector('book-source', 'book', slow=True)
    (EVIDENCE/'connector-config.json').write_text(json.dumps(cfg, indent=2))
    def active_scan():
        return [r for r in sql(source, 'SHOW FULL PROCESSLIST') if r.get('Info') and 'SELECT * FROM inventory.products WHERE SLEEP' in r['Info']]
    before = eventually(active_scan)
    timestamp_before = sql(source, 'SELECT UTC_TIMESTAMP(6) AS t')[0]['t'].isoformat()
    sql(source, "INSERT INTO inventory.products VALUES (3001,'snapshot-concurrent-insert',321,1,1)")
    sql(source, "UPDATE inventory.products SET price_cents=777,revision=2 WHERE id=1")
    sql(source, 'DELETE FROM inventory.products WHERE id=2')
    timestamp_after = sql(source, 'SELECT UTC_TIMESTAMP(6) AS t')[0]['t'].isoformat()
    after = active_scan()
    overlap = {'before': before, 'after': after, 'write_start_mysql_utc': timestamp_before,
               'write_end_mysql_utc': timestamp_after, 'write_ids': [3001,1,2]}
    (EVIDENCE/'snapshot-write-overlap.json').write_text(json.dumps(overlap, default=str, indent=2))
    check('writes occurred while real snapshot SELECT remained active', bool({r['Id'] for r in before} & {r['Id'] for r in after}), overlap)
    first = worker('main', 'live', 'book.inventory.products')
    (EVIDENCE/'initial-consume.json').write_text(json.dumps(first, indent=2))
    check('real snapshot and binlog r/c/u/d captured', all(first['ops'].get(op,0)>0 for op in ('r','c','u','d')), first['ops'])
    check('real binlog file and position captured', any(p.get('file') and p.get('pos') for p in first['source_positions']))
    reconcile(source, pg, 'live', 'snapshot-plus-binlog')
    replay = worker('main', 'live', 'book.inventory.products', ['--replay'])
    check('duplicate replay detected in transactional inbox', replay['duplicates'] == first['messages'] and replay['fresh'] == 0, replay)
    reconcile(source, pg, 'live', 'duplicate-replay')
    before_invalid = (pg.execute('SELECT count(*) FROM products_projection').fetchone()[0], pg.execute('SELECT count(*) FROM delivery_history').fetchone()[0])
    rejections=[]
    for group,generation,topic in [('main','forbidden-new','book.inventory.products'),('other','live','book.inventory.products'),('main','live','wrong-topic')]:
        bad=subprocess.run([sys.executable,str(LAB/'consumer.py'),'--group',group,'--generation',generation,'--topic',topic],capture_output=True,text=True,timeout=15)
        rejections.append({'group':group,'generation':generation,'topic':topic,'returncode':bad.returncode,'explicit_binding_error':'Generation must bind' in bad.stderr})
    after_invalid = (pg.execute('SELECT count(*) FROM products_projection').fetchone()[0], pg.execute('SELECT count(*) FROM delivery_history').fetchone()[0])
    check('generation group topic mismatch rejected without pollution', all(r['returncode']!=0 and r['explicit_binding_error'] for r in rejections) and before_invalid==after_invalid and pg.execute('SELECT count(*) FROM consumer_binding').fetchone()[0]==1, rejections)
    # Fresh group: one durable DB transaction, then forcibly kill the real process
    # while its code is blocked before consumer.commit().
    marker = EVIDENCE/'kill-ready.json'
    command = [sys.executable, str(LAB/'consumer.py'), '--group', 'crash', '--generation', 'crash-gen',
               '--topic', 'book.inventory.products', '--crash-after-db', '--ready-file', str(marker)]
    with (EVIDENCE/'crashed-process.log').open('w') as log:
        child = subprocess.Popen(command, stdout=log, stderr=log)
        eventually(lambda: marker.exists(), 60)
        position = json.loads(marker.read_text())
        check('database committed before process kill', pg.execute('SELECT count(*) FROM delivery_history WHERE consumer_group=%s', ('crash',)).fetchone()[0] == 1)
        probe = Consumer({'bootstrap.servers':'127.0.0.1:29092','group.id':'crash','enable.auto.commit':False})
        committed = probe.committed([TopicPartition(position['topic'],position['partition'])], timeout=10)[0].offset
        probe.close()
        check('broker offset uncommitted at kill barrier', committed < 0 or committed <= position['offset'], {'broker_next_offset':committed, **position})
        child.kill()
        child.wait(timeout=10)
        check('real subprocess terminated by SIGKILL', child.returncode == -signal.SIGKILL)
    restarted = worker('crash', 'crash-gen', 'book.inventory.products')
    check('restart re-delivers committed-database event', restarted['duplicates'] >= 1, restarted)
    reconcile(source, pg, 'crash-gen', 'crash-restart')
    # No consumer runs while these real source transactions create backlog.
    for i in range(100):
        sql(source, 'UPDATE inventory.products SET price_cents=price_cents+1,revision=revision+1 WHERE id=%s', (10+i,))
    sql(source, 'DELETE FROM inventory.products WHERE id=10')
    sql(source, "INSERT INTO inventory.products VALUES (4001,'after-outage',222,1,1)")
    stale = source_rows(source) != projection_rows(pg, 'live')
    check('stopped consumer leaves projection observably stale', stale)
    caught = worker('main', 'live', 'book.inventory.products')
    check('backlog drained after consumer restart', caught['ops'].get('u',0)>=100 and caught['ops'].get('d',0)>=1, caught)
    reconcile(source, pg, 'live', 'backlog-catchup')
    sql(source, "ALTER TABLE inventory.products ADD COLUMN note VARCHAR(80) NULL")
    sql(source, "UPDATE inventory.products SET note='new schema field',revision=revision+1 WHERE id=1")
    # Existing untouched rows receive the schema default on rebuild; live schema
    # migration normalizes the same null default before reading new events.
    pg.execute("UPDATE products_projection SET document=document || '{\"note\":null}'::jsonb WHERE generation='live'")
    evolved = worker('main', 'live', 'book.inventory.products')
    check('real ALTER and subsequent binlog update consumed', evolved['ops'].get('u',0)>=1)
    check('added field reaches query document', pg.execute("SELECT document->>'note' FROM products_projection WHERE generation='live' AND id=1").fetchone()[0] == 'new schema field')
    reconcile(source, pg, 'live', 'schema-evolution')
    connector('book-rebuild', 'rebuild', slow=False)
    rebuilt = worker('rebuild', 'new-generation', 'rebuild.inventory.products')
    check('isolated new snapshot rebuild consumed', rebuilt['ops'].get('r',0)==len(source_rows(source)), rebuilt['ops'])
    reconcile(source, pg, 'new-generation', 'full-rebuild')
    check('live generation preserved during rebuild', projection_rows(pg,'live') == projection_rows(pg,'new-generation'))
    statuses = {name: request('/connectors/'+name+'/status') for name in ('book-source','book-rebuild')}
    check('both actual connectors and tasks RUNNING', all(s['connector']['state']=='RUNNING' and len(s['tasks'])==1 and all(t['state']=='RUNNING' for t in s['tasks']) for s in statuses.values()), statuses)
    result = {'status':'PASS','skipped':0,'checks':checks,'check_count':len(checks),'duration_seconds':round(time.time()-start,3),
              'versions':versions,'run_url':os.environ.get('BOOK_RUN_URL'),'source_revision':os.environ.get('BOOK_SOURCE_SHA'),
              'boundaries':['single Kafka broker; one data partition','real Debezium initial snapshot and MySQL binlog',
                            'local teaching credentials only','no multi-node failover or binlog-loss proof',
                            'cross-store delivery is at-least-once with projection convergence; not end-to-end exactly-once']}
    (EVIDENCE/'runtime.json').write_text(json.dumps(result, ensure_ascii=False, indent=2))
    print(json.dumps({'status':'PASS','checks':len(checks),'seconds':result['duration_seconds']}))


if __name__ == '__main__':
    main()

lab/run.sh

#!/bin/sh
set -eu
cd "$(dirname "$0")/.."
: "${CDC_COMPOSE_PROJECT:?Set a dedicated CDC_COMPOSE_PROJECT before running; this lab owns only the dedicated project resources}"
case "$CDC_COMPOSE_PROJECT" in cdc-book-*) ;; *) echo 'Project must start cdc-book-' >&2; exit 2;; esac
mkdir -p "${CDC_EVIDENCE_DIR:-lab/evidence}"
if [ -f "${CDC_EVIDENCE_DIR:-lab/evidence}/kill-ready.json" ]; then
  echo "Refusing stale crash marker; use a fresh evidence directory" >&2
  exit 2
fi
cleanup() {
  docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml logs --no-color > "${CDC_EVIDENCE_DIR:-lab/evidence}/containers.log" 2>&1 || true
  docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml down --volumes --remove-orphans
}
trap cleanup EXIT
python3 -m pip install -r lab/requirements.txt
docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml up -d
for task_service in mysql kafka connect postgres; do
  task_container=$(docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml ps -q "$task_service")
  task_image=$(docker inspect --format '{{.Image}}' "$task_container")
  docker image inspect --format '{{json .RepoDigests}}' "$task_image" > "${CDC_EVIDENCE_DIR:-lab/evidence}/image-$task_service.json"
done
docker compose -p "$CDC_COMPOSE_PROJECT" -f lab/compose.yml exec -T kafka /opt/kafka/bin/kafka-topics.sh --version > "${CDC_EVIDENCE_DIR:-lab/evidence}/kafka-version.txt"
python3 lab/run.py

lab/schema.sql

CREATE DATABASE inventory;
CREATE USER 'cdc'@'%' IDENTIFIED BY 'cdc-book-local';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT, LOCK TABLES ON *.* TO 'cdc'@'%';
CREATE TABLE inventory.products (
  id BIGINT PRIMARY KEY,
  sku VARCHAR(64) NOT NULL,
  price_cents INT NOT NULL,
  active TINYINT NOT NULL,
  revision INT NOT NULL
);