# 前言：把同步结果变成可以证明的事实 业务系统通常先拥有一张可靠的交易表，随后出现全文搜索、运营筛选、知识检索和分析报表。团队很快发现，把同一行再复制一份并不困难；困难在于原表继续变化、消费者停止、旧数据删除、字段上线之后，还能回答那份副本究竟代表哪个时刻的事实。本书关注的对象就是这种持续变化的数据投影。 我们用一个小型商品目录贯穿全书：MySQL 保存业务事实，Debezium 读取真实二进制日志，Kafka 保留变化事件，Python 消费者把事件写入 PostgreSQL 查询投影。选择 PostgreSQL 是为了让事务、唯一约束和完整对账清晰可见；这里的投影不是全文搜索引擎，也没有声称覆盖 Elasticsearch 的分词、分片与排名行为。读者可以在理解同步边界后替换查询存储，不能直接把本书的数据库事务保证搬到另一种接口上。 所有集成实验都要求四个真实服务就绪，依赖缺失会失败。我们在初始快照查询执行期间插入、修改和删除数据，随后观察真实快照事件与 binlog 事件。我们实际杀死投影已提交、Kafka 位点尚未提交的消费者进程，验证重启产生重复投递；再演练停机积压、新增字段、隔离全量重建与逐行核对。随书报告记录检查项、服务版本、镜像实际摘要与运行地址，失败的报告不会被包装为已完成验证。 本书假设读者理解 SQL 的基本查询和数据库事务，能够运行命令行程序，愿意阅读少量 Python。前六章建立边界，第七至十二章解释在线投影，第十三至十八章完成重建、运维和交付。每章练习都有参考思路。完整代码自动从实验目录编入附录，下载 ZIP 可以在独立目录重新执行，不需要先克隆整套出版工程。 需要提前约定一个词：可靠并不等于永不出错。这里的可靠是能够保留必要证据，在限定条件下重试、恢复并验证结果。单个 broker、单个数据分区与合成数据无法证明多机容灾或大规模容量；个人学习凭据也不能直接用于生产。我们会把这些限制写在结论旁边，让实验支持的结论与尚待验证的设计保持一致。 # 第1章 同步契约：先定义副本代表什么 ## 从一次搜索投诉开始 商品价格已经在管理后台改成七百七十七分，搜索页却仍显示旧值。值班者看到连接器状态正常、Kafka 有流量、消费者也没有报错，仍然不知道问题在哪里。这不是缺少一个更大的监控面板，而是系统没有把正确性拆成可以检验的契约。源表写入成功只说明 MySQL 接受了事务；连接器正常只说明任务当前在运行；消费位点前进也不能单独证明查询对象已经更新。 我们的第一个契约是身份。商品由主键 id 唯一标识，sku 是可变化属性，不能同时把它当成投影永久身份。第二个契约是内容：投影保存 id、sku、price_cents、active、revision，后续演进增加可空 note。价格用整数分，避免对账时把浮点表示误差误判成同步失败。第三个契约是删除：源表物理删除后，在线查询不能返回该商品；投影内部可以保留删除标记用于阻止旧事件复活。 ![CDC同步边界与恢复证据](../images/pipeline.svg) ## 把时间分成不同边界 源库事务提交、连接器读取日志、Kafka 接收记录、消费者提交投影、消费者提交位点是五个不同的时刻。把它们统称为“同步完成”会掩盖故障窗口。业务接口若需要读到刚写入的值，可以返回源库确认，或者等待一个明确的投影水位；仅等待固定两秒不能提供读己之写保证。本文实验最终停止写入并完整对账，证明该批有限变更已收敛，不能把这个结果表述成持续写入时的线性一致查询。 我们把可接受的数据延迟看成服务目标，把最终内容一致看成正确性目标。延迟高可能仍然正确，内容错误即使非常快也不能接受。例如消费者把 price_cents 当成元展示，所有消息都会成功处理，却破坏业务语义。再如删除事件被过滤，插入和修改都正常，查询副本仍保留已经不存在的商品。正确性必须覆盖操作种类和字段语义，而不是只统计消费数量。 ## 投影不是第二个事实源 MySQL 是本书商品事实的唯一写入入口。投影数据库由同步程序维护，业务界面不能直接修改它再期待变化反向传播。否则两个存储都能产生事实，需要冲突检测、来源标识和优先级规则，这已经是双向同步问题。我们刻意维持单向链路，使重建能从源库出发得到确定结果。 这种选择也决定恢复方式：损坏的查询投影可以丢弃并重建，源库不能因为投影异常而被反向覆盖。删除标记、消费历史和检查点属于投影的恢复元数据，并不是新增的业务事实。它们可以帮助判断哪些事件已经应用，却不能单独恢复被源库与日志共同丢失的业务变更。因此可重建不等于免备份，日志保留也不等于完整灾备。 ## 正确性检查需要覆盖什么 在实验里，对账按照主键排序读取源表全部行，读取投影里所有未删除行，比较完整 JSON 字段和值。我们同时记录行数和规范序列化后的摘要，但通过条件是实际列表相等。仅行数相同无法排除一行缺失、一行重复；仅抽样主键无法排除价格错位；仅比较最大 revision 无法排除中间对象丢失。小数据允许完整比较，生产中可以分桶与分段执行相同逻辑。 正确性还包括变更期间的行为。新字段上线时，旧对象没有收到更新事件，投影也必须有明确的默认值语义。删除后的旧事件重放不能复活对象。消费者崩溃后，重放不能重复扣款之类的业务副作用。本书投影只更新查询文档，不执行支付操作；若读者把消费者改成外部命令处理器，需要重新建立业务幂等协议，不能只复用当前表结构。 ## 把失败语义写进接口 查询接口也应当告诉使用者数据可能落后，而不是在后台积压时继续展示毫无上下文的旧值。对于普通商品目录，可以接受短暂延迟并保留源库管理入口；对于余额、库存扣减或访问许可，延迟副本可能不足以作最终决策。同步架构应由业务风险决定读取路径，不能先搭好流式管线再让所有接口迁入。契约还应明确字段来源：哪些值来自权威表，哪些是投影计算，哪些只是展示缓存。否则对账发现差异时，团队会争论究竟该修源还是修副本。 ## 本章练习 给一份用户目录定义同步契约，至少写出身份字段、可变字段、删除含义、可接受延迟和对账方法。再回答：用户禁用与物理删除是否相同？如果查询系统承担权限过滤，延迟期间是否允许继续展示旧结果？这两个问题的答案会直接影响后续事件模型和读取策略。 # 第2章 日志边界：理解 CDC 能证明什么 ## 为什么从日志读取 定时查询 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 已有事件而投影没有更新，列出至少三种可能原因，并说明每种原因需要读取什么证据来区分。 # 第3章 实验环境：让证据可以重新产生 ## 四个真实服务与一个消费者 随书环境使用 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 和卷清理，不扫描或删除机器上所有容器。固定主机端口使命令容易阅读，也意味着同一主机不能同时运行两份实验；端口冲突应当失败，不能通过杀死未知进程解决。项目名前缀不是权限隔离机制，只是防止误操作的明确约束。 ```sh 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 插件版本和四个镜像摘要，说明它们各自证明了什么。然后故意占用一个实验端口，确认实验明确失败且没有清理其他项目的资源。 # 第4章 业务模型：选择能够解释的变化 ## 一个足够小但不失真的目录 实验源表 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 作为键，会在哪些情况下覆盖另一租户的数据。 # 第5章 初始快照：在数据继续变化时开始同步 ## 第一次接入为什么需要快照 数据库已经存在两千行，连接器此时才开始读取日志。如果只订阅未来变化，长期未更新的商品永远不会进入查询系统。初始快照解决这个起点问题：读取一份已有数据，并把后续日志接到这份数据之后。真正困难的地方是读取期间业务不能长期停止，快照与日志必须围绕一致性边界协调。 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 后再运行实验，解释为什么交叠检查可能失败，且这个失败不能证明连接器快照算法错误。再设计一种不依赖固定等待秒数的替代同步点，写出它能够证明的边界。 # 第6章 快照与增量衔接：把水位放回命名空间 ## 一个位置需要带上身份 “当前位点是一千”几乎没有诊断价值。至少应当知道它属于哪个源实例、哪个日志文件，或者哪个 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 十和二十，说明为什么不能声称全局水位为二十。再列出把分区数从一改为三之前必须验证的键路由、旧数据、投影条件和切换步骤。 # 第7章 消息结构与键：保持对象身份稳定 ## 先看真正到达客户端的字节 连接器文档展示的逻辑事件，与 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 的消息。 # 第8章 投影幂等：让重复交付成为普通情况 ## 重复并不是例外路径 只要投影数据库和 Kafka 位点不处于同一个原子事务中，就存在数据库成功、位点尚未确认的窗口。进程在窗口内退出，重启后 broker 会再次投递该记录。把重复视为罕见脏数据，然后依靠人工删除，会让正常恢复变成危险操作。我们先承认至少一次交付，再让重复处理保持业务结果不变。 幂等的对象必须具体。把同一商品文档设置成相同值，通常可以幂等；把价格每次加一则不可以。随书消费者使用事件 after 的完整值覆盖投影，并使用历史唯一约束阻止同组坐标再次执行更新。若以后需要累计统计，应当保留事件级去重并在同一事务内更新计数，不能只依靠文档 upsert 的经验。 ## 一次事务包含三件事 处理记录时，程序开启 PostgreSQL 事务，先向 delivery_history 插入事件身份。插入成功才应用业务变化；若唯一约束冲突，说明该消费组已经处理过该记录，跳过业务更新。随后推进数据库中的 delivery_checkpoint，最后共同提交。历史记录、查询文档和数据库检查点的可见性属于同一个数据库事务。 如果更新投影失败，事务回滚，历史插入也回滚，下一次投递仍能正常尝试。如果先单独提交历史，再更新投影，进程可能在中间退出，重启时看到历史存在而永远不再补投影。幂等表的存在本身不提供保证，关键在于它与实际状态变化共享提交边界。 数据库事务提交以后，程序再同步提交 Kafka 位点。此时第二个系统的操作失败会产生重复，不会让尚未持久化的投影被提前确认。反过来先提交位点再写数据库，会在中间崩溃时丢失投影更新，因为消费组已越过该记录。顺序决定主要故障结果：我们选择可重试的重复，避免无证据的跳过。 ## offset 条件是第二层防线 历史去重适用于同一消费组的重复坐标。显式回放或不同恢复场景还需要防止旧事件把新文档覆盖。投影更新条件要求分区相同且新 offset 更大。因此即使尝试重放旧创建事件，已经应用的删除标记不会倒退成未删除状态。 这个条件依赖固定 topic 和键路由，所以程序启动时验证绑定关系。新 topic 的 offset 从零开始，不能直接与旧 topic 的大数比较；换分区后的数字也没有大小关系。把这些限定写成数据库绑定和启动拒绝，比只在 README 中提醒使用者更可靠。错误组合会在消费之前退出。 ## 真实重复实验怎样验收 首轮消费结束后，控制程序用同一消费组、同一世代显式从分区起点重放。这个操作只用于实验，生产中不能随意改变运行组的位点。消费者报告重复数等于首轮消息数，新鲜处理数为零；随后再次读取源库与投影，要求全部行与字段仍然相等。 我们还测试三种错误组合：旧消费组用于新世代、同一世代换消费组、同一世代换 topic。每次启动都必须明确报绑定错误，投影行数、历史行数和绑定数量保持不变。这样证明参数拒绝发生在数据处理前，而不是等错误结果出现再靠对账发现。 ## 幂等历史也有生命周期 永久保存每个事件坐标会增长存储，删除历史又可能使非常旧的重复再次进入业务逻辑。对纯当前状态投影，offset 条件仍提供一部分保护，但涉及累加、外部副作用或跨对象操作时，不能随意缩短去重窗口。清理策略应与可回放范围、日志保留和重建方案一起设计。 本实验不实现历史清理，因为有限数据允许完整保留。它也不实现跨系统支付幂等或外部通知去重。若消费者要调用远程服务，需要远端接受稳定业务幂等键，并明确未知结果查询协议；本地历史事务不能原子包含一个任意 HTTP 调用。 ## 本章练习 把投影更新想象成一次积分加一，说明只用 ON CONFLICT 更新对象为什么不够。然后画出“历史已提交、投影未提交”和“投影已提交、位点未提交”两种故障窗口，解释哪一种被当前事务边界消除了。 # 第9章 位点前崩溃：用真实进程验证恢复 ## 把故障放在明确的提交窗口 测试消费者恢复时，随便启动进程再立刻 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 尚未确认的重复恢复路径。 # 第10章 消费停机与积压：恢复速度必须快于变化速度 ## 先区分是谁停止了 业务源库继续写入时，消费者可以停止，连接器也可以停止，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 数据过期，相关策略属于设计要求。 ## 结束恢复需要业务验收 当积压接近零，还应检查删除对象是否消失、新插入对象是否存在、关键字段是否正确。我们的逐行核对覆盖所有商品，生产可以先运行重要对象探针，再分段进行全量对账。只有交付进度和内容检查都满足要求，才能结束恢复事件，避免下一次用户查询才发现残留错误。 ## 重试应当有可观察的代价 投影暂时不可用时重试是合理动作，但重试速率需要受控。无限快速重连会消耗数据库连接和日志，也会挤占同机其他业务。生产客户端应区分暂时故障与契约错误，使用退避和告警；解析不支持事件不应每秒重启一次而永远不通知维护者。本书有限集成遇错即失败，便于留下清楚现场，尚未实现长期运行的退避控制器。扩展前应保留这种错误可见性。 ## 本章练习 假设停机一小时产生三十六万条积压，持续输入每秒一百条，恢复能力每秒三百条，估算理想追平时间。随后列出至少四个会让实际时间更长的条件，并说明哪些条件可以通过增加消费者实例解决。 # 第11章 删除与可见性：防止旧对象重新出现 ## 删除比更新更容易被忽视 插入和更新都有 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”不能作为投影不会复活旧对象的充分证据。 # 第12章 Schema 演进：字段出现不等于语义已经迁移 ## 一个可空列带来的两种状态 源表最初没有 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 的迁移条件，解释它为何比无条件覆盖更适合持续消费场景。 # 第13章 隔离重建：为查询副本建立新世代 ## 为什么恢复不能只有续读 消费位点仍然存在，未必说明当前投影值得继续使用。早期消费者可能遗漏删除，字段转换可能有错误，运维也可能误改查询数据。继续消费只能应用未来变化，无法自动修复所有历史偏差。可靠系统需要一条从权威源重新生成查询状态的路径，使正确性不依赖对旧故障的完整猜测。 重建不应直接清空在线投影。清空后慢慢补数据会让用户看到大量缺失，失败时也没有可回退版本。我们使用 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 组状态和历史表空间。当前实验只在隔离容器结束时整体清理，没有提供生产世代回收调度器。 ## 重建也要考虑源库压力 重建连接器会再次扫描源表，并与在线连接器共同消耗资源。大型业务不能因为查询副本“可丢弃”就无限重复重建。需要估算扫描时间、额外日志生成、快照事务存活、网络带宽与投影写入成本。失败后从头重建是否可接受，也应当在服务恢复目标中有明确答案。 本书通过两千行和完整比较建立方法，不提供大表重建耗时承诺。生产可分租户、分范围建立世代，或者选择增量快照，但每次拆分都要保留集合覆盖和删除处理的证明。只把全表拆成多个任务，没有共同水位，可能形成一个从未真实存在过的混合视图。 ## 重建完成后还要验证读取路径 数据在新世代中正确，并不代表用户已经读到它。查询配置可能仍然指向旧世代，缓存可能持有旧结果，导出任务也可能绕过统一入口。生产切换验收应使用实际查询接口读取已知新增、更新和删除对象，再核对世代标识。当前实验只比较数据库文档，没有一个生产商品查询服务，因此不会把数据库内的新世代相等称为端到端线上切换已经完成。 ## 本章练习 为持续写入的商品系统设计世代切换协议，写出快照完成、增量追平、验收、指针切换、观察与回收六个边界。再指出本书哪些边界已经有真实实验，哪些仍需要你新增验证。 # 第14章 完整对账：证明内容相等而非指标相近 ## 从行数开始，但不要停在那里 源表和投影都显示两千行，看起来令人放心。实际上源表可能有 id 一到两千，投影却缺少 id 二、多出一个已删除对象，行数仍然相同。再加上字段错位、价格单位和 null 语义，单纯数量比较远远不足。行数适合作为快速探针，不能作为最终正确性证明。 本书从源库按 id 排序读取全部业务字段，从投影读取相同世代里未删除的 document，并按 id 排序。实际通过条件是两个完整列表直接相等。随后才把规范 JSON 序列化后的 SHA256 写入报告，便于跨运行核对。摘要帮助压缩证据，却不替代程序真正执行的内容比较。 ## 比较前先统一语义 整数价格、整数 active 与 revision 在两端保持相同类型。新增 note 后，源库旧行的 null 必须在投影中表现为同样键和值。若一边没有键，另一边键为 null，当前契约会判为不同。这种严格性有助于发现 schema 迁移遗漏，也要求我们先定义什么叫等价。 真实生产可能存在时区、精度、排序、字符归一化和默认值问题。对账程序应共享业务契约，而不是共享消费者的全部实现。如果两者复用同一个错误转换函数，可能一起把错误当正确。实验源侧直接读取数据库行，投影侧直接读取文档，没有用消费事件重新构造源侧预期，从而减少这种同源错误。 ## 六次比较覆盖不同阶段 初始快照加增量比较验证接入衔接；重复回放比较验证幂等；崩溃重启比较验证重复恢复；停机追平比较验证积压；schema 演进比较验证新增字段与默认值迁移；全量重建比较验证新世代。每次都保存单独报告，使失败可以定位到某一阶段，而不是只在最后发现所有结果都不对。 对账也会检查删除后的结果，因为源库不存在的对象不会出现在源列表，而投影查询固定排除 deleted。若消费者没有删除标记，或者查询忘记过滤，列表长度或具体 id 都会不一致。这个检查覆盖业务可见性，但不会检查内部删除文档是否已经从磁盘擦除，那是不同目标。 ## 对账时的时间参照 两个数据库之间没有共同读事务。如果源库在读取过程中继续变化，而投影在另一个时间被读取，结果差异可能来自时间边界，而不是永久错误。实验为有限批次停止写入后进行比较，因此参照稳定。生产持续系统需要固定水位、快照、变更窗口或允许重试的差异分类，不能每次比较都把短暂延迟当成数据损坏。 一种做法是按对象业务版本判断是否已经可比较：只核对源版本不再变化并且投影已经达到相应版本的对象。另一种是建立一致快照与日志截止边界。这些方法需要来源支持，不能简单取两个系统各自的当前时间。时钟同步有帮助，但不能把独立提交变成原子提交。 ## 大规模对账如何扩展 数据量大时可以按稳定主键范围分段，记录每段行数、键集合摘要和字段摘要，先定位差异段，再逐行钻取。分段必须防止范围遗漏和重叠，不能只比较随机样本。哈希摘要还应包含字段名称和明确的编码规则，避免拼接歧义；例如两个字符串字段直接连接会让不同元组产生相同字节串。 本书使用结构化 JSON、固定排序与明确字段，因此可重复生成摘要。SHA256 相同仍是基于哈希抗碰撞假设的压缩证据，实际实验还比较原始列表。生产是否保留逐行明细，要结合敏感数据和证据体积决定，可以保留差异键与脱敏字段，而不是把整库内容上传到公开 CI。 ## 发现差异之后不要立即补丁 先保留源快照、投影状态、最后交付位置与相关消息，再判断差异类型。少量对象落后可能是延迟，永久缺失可能是位点跳过，旧值覆盖可能是键路由或顺序错误。直接在投影上手工修复虽然能让一次比较通过，却可能被之后重放再次覆盖，也可能掩盖系统性错误。 修复应当对应原因：消费者逻辑有误则发布修正并重建，个别消息失败则按保留的输入安全重试，源数据错误则在权威源修复。对账是定位与验收工具，不应变成绕过同步协议的第二写入入口。 ## 差异报告要避免泄露源数据 公开CI适合合成数据，不适合把真实客户行直接上传artifact。生产对账可以记录差异主键、字段名、版本和受控摘要，把敏感原值留在受权限保护的系统里。摘要也可能受到低熵字段枚举推测，不能自动当成脱敏。证据设计需要同时满足定位与最小暴露，两者都应在上线前确定，而不是出故障后把整库导出到临时公共位置。 ## 本章练习 构造两份行数相同但内容不同的数据，说明数量和最大 id 为什么都会通过。再写一个分段对账方案，明确边界、规范编码、差异明细和持续写入期间的重试条件。 # 第15章 故障定位：沿提交边界寻找第一处分歧 ## 让排查从问题对象开始 用户报告 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，使检查结果能够回到具体源码提交和运行环境。公开报告只使用合成数据，因此可以完整分享；生产报告需要限制敏感字段，但仍应保留足够的事件身份来重现判断过程。 ## 本章练习 为“源库已删除，查询仍返回对象”写一份最短排查路径。每一步列出读取证据、可能结果和下一步，禁止用“重启试试”代替判断。再说明在哪一步可以区分连接器没发删除与查询忘记过滤。 # 第16章 容量与延迟：为恢复保留余量 ## 先计算哪些数据会增长 源库表会增长，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、内存、连接、队列和磁盘预算，并在达到阈值时停在可恢复位置。单纯把超时设得很大，只会延后暴露资源耗尽。增加这些限制时也要测试失败路径，确认不会在投影尚未提交时推进位点；资源保护只有与正确性协议一起工作，才真正提高可恢复性。 ## 本章练习 按每天一千万条变更、每条一千字节、保留三天估算原始日志量，再列出未包含的副本、索引、压缩和峰值因素。为当前单分区消费者设计一项批处理优化，写明必须保留的事务与位点顺序。 # 第17章 上线与迁移：把可重建能力变成交付流程 ## 上线不是把 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。 # 第18章 恢复演练：把系统保证逐项验收 ## 用一条完整时间线串起全书 实验首先创建源表和两千行合成商品，初始化空投影。连接器开始真实 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可以帮助验证局部分支，却不能替代复制日志已经丢失时的集成恢复证明。每项演练都应写清故障触发点、保留证据和适用边界。 # 附录二 官方来源与阅读方法 本书使用下列一手文档核对概念和配置。作者实际抓取的页面摘要、地址和SHA256见下载区sources.json及仓库research/sources.md。完整第三方页面仅保存在被忽略的本地缓存，不随仓库或电子书再分发。页面内容可能随官方维护改变，抓取摘要标识的是当次查阅内容，不代表永久版本。 - [Debezium 3.1 MySQL连接器](https://debezium.io/documentation/reference/3.1/connectors/mysql.html)：初始快照、日志配置、schema历史和删除包络。书中实验只覆盖明确列出的配置。 - [Debezium topic自动创建](https://debezium.io/documentation/reference/3.1/configuration/topic-auto-create-config.html)：连接器数据topic的创建参数，帮助解释分区与副本设置。 - [MySQL 8.4复制格式](https://dev.mysql.com/doc/refman/8.4/en/replication-formats.html)：区分行、语句和混合日志，理解ROW要求。 - [MySQL 8.4二进制日志](https://dev.mysql.com/doc/refman/8.4/en/binary-log.html)：日志用途、保留与位置边界，避免把日志当作无限备份。 - [Kafka 3.9文档](https://kafka.apache.org/39/design/design/)：分区顺序、消费位点和交付语义。Kafka事务保证不能自动扩展到任意外部数据库。 - [Confluent Python客户端API](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html)：Consumer轮询、同步提交、已提交位点与管理接口。抓取文档为2.16.0，实验依赖固定为2.10.0，所用API由真实实验核验。 - [PostgreSQL 17 INSERT](https://www.postgresql.org/docs/17/sql-insert.html)：ON CONFLICT与条件更新，支撑事件唯一约束和投影防倒退。 - [PostgreSQL 17事务隔离](https://www.postgresql.org/docs/17/transaction-iso.html)：数据库内可见性边界。单库事务不能被表述为跨MySQL、Kafka、PostgreSQL的全局事务。 阅读官方文档时，应先确认版本，再区分默认值、可选行为与实验配置。若官方支持某能力，而随书代码没有实现和验证，就不能把那项能力列为本书实验结论。相反，真实报告通过也只支持记录的版本、环境和故障范围。 # 附录三 术语与保证边界 **CDC**：捕获数据库变化的技术集合。本书特指Debezium读取MySQL真实binlog并输出事件的路径，不把定时扫描updated_at当作相同协议。 **快照**：在既定一致性边界下读取已有数据。快照完成不等于投影已追平；本实验在扫描期间真实写入，再验证最终集合。 **binlog位置**：MySQL日志命名空间内的位置，不能直接与Kafka offset比较，也不能跨不相关源实例取数值最大值。 **Kafka offset**：某个topic分区内的记录位置。消费组提交的通常是下一个待读位置，当前记录零与下一位置一不是同一含义。 **至少一次交付**：故障恢复可能重复投递。本文用数据库内去重和条件更新处理重复，不称为端到端恰好一次。 **幂等**：重复相同操作不改变结果。必须明确操作身份和持久化边界；一张去重表不自动保证远端副作用幂等。 **tombstone**：同键空值Kafka记录，用于日志压缩。它不同于带删除信息的Debezium d包络，也不同于投影内deleted标记。 **世代**：一次投影构建的隔离身份。本实现要求世代固定绑定一个唯一消费组和topic，新建世代不能复用旧组历史。 **对账**：按明确业务语义比较源与投影。本文通过条件是完整行列表相等，行数与SHA256只是辅助证据。 **重建**：从权威源和明确输入重新生成派生状态。本文验证隔离全量快照与完整内容，不声称已实现持续写入时生产查询的原子切换。 **Schema迁移**：字段与文档结构的版本变化。新增源列不会自动为历史投影回填字段，本实验明确执行投影默认值迁移。 **恢复窗口**：日志、状态与容量允许恢复的范围。任何清理都应与可回放位置和重建策略共同设计。 本书所有数据为合成商品，不执行支付、用户身份认证或权限撤回。单broker、单分区与同机容器实验不证明多节点高可用、断电持久性或跨机房容灾。 # 附录：随书实验完整代码 以下文件在构建时从随书实验与测试目录读取，与下载区实验包同源。它们是本书教学实现；实际验证范围以实验章节和验证记录为准。 ## lab/README.md ~~~~~~text # 真实 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 ~~~~~~yaml 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 ~~~~~~python #!/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 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 ~~~~~~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 ~~~~~~text PyMySQL[rsa]==1.1.1 confluent-kafka==2.10.0 psycopg[binary]==3.2.9 ~~~~~~ ## lab/run.py ~~~~~~python #!/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 ~~~~~~bash #!/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 ~~~~~~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 ); ~~~~~~