机制与取舍：分布式集群怎么协调自己

五个机制按从控制面到数据正确性的顺序排列：分布式协调（集群怎么不靠外部协调器活着）→ 再平衡（拓扑变了怎么重分 task）→ 序列化机制（字节层如何编解码）→ 偏移管理与恢复（重启从哪续）→ EOS source（不重不漏怎么做到）。每个机制先讲运行方式，再用备选方案对比表说清"为什么没选别的设计"，然后是代价与失效模式，最后一道预测题。01 章给的是词汇表，这一章给的是这些词背后的工作机制与取舍——这是从"会配"到"理解"的那一层。

2.1分布式协调：控制面全在 Kafka 自己的 topic 里

distributed 集群不靠任何外部协调器协调，而是用 Kafka 的消费组协议加上三个 compacted internal topic——控制面就是 Kafka 本身。

运行方式

多个 worker 配同一个 group.id，它们通过 Kafka 的消费组成员协议（与普通消费者加入消费组用的是同一套 group membership 机制）加入同一个 Connect 集群。集群里有一个被选出的 leader worker 负责计算 task 分配方案，其余 worker 执行分到自己头上的 task。三类需要持久化、需要跨 worker 一致的状态，各写进一个 compacted topic：

• connect-configs：connector 与 task 的配置。POST 一个 connector 的本质，就是往这个 topic 追加一条配置记录。
• connect-offsets：source connector 的偏移（见 §2.4）。
• connect-status：connector 与 task 的运行状态（RUNNING/FAILED/PAUSED），GET /status 读的就是它。

这三个 topic 全是 compacted（日志压缩）——压缩保留每个 key 的最新值，正好匹配"配置 / 偏移 / 状态都只关心当前值"的语义。REST API（§1.2 已见）不是一个独立服务，它只是这套机制的门面：一次管理请求落到任意 worker，被翻译成对这三个 topic 的读或写，集群里所有 worker 读到后协同执行。这就是 01 章那句"配置即数据"在机制层的展开。

备选方案对比

"集群协调"是分布式系统的经典难题，常规答案是引入一个外部协调器（ZooKeeper、etcd）。Connect 选了另一条路。

代价与失效模式

"控制面就是几条 topic"换来零外部依赖，代价是这几条 topic 的健康直接等于集群的健康。两种典型失效：

• internal topic 配置 / 压缩不一致 = 裂集群。如果新加的 worker 配了不同的 internal topic 名，或这些 topic 没设成 compacted、副本数不足，集群会读不到一致的控制面，表现为 connector 莫名消失、配置回滚、状态错乱。
• group.id 配错 = 一个集群裂成两个。两批本应同属一个集群的 worker 配了不同 group.id，会各自组成独立集群、各跑一份 connector——同一个 source 被搬两遍。

与下一个机制的关系：worker 靠消费组协议加入集群——那么一个 worker 加入或离开时，它名下的 task 怎么重新分配给其他 worker？这就是再平衡，而 Connect 的再平衡方式经历过一次关键演进。

2.2再平衡：incremental cooperative 只暂停被移动的 task

再平衡是 worker 加入 / 离开 / 配置变更时重新分配 task 的过程；自 KIP-415（2.3）起用 incremental cooperative 策略，只暂停被收回或移动的那部分 task，其余继续搬。

运行方式

集群拓扑变化（一个 worker 宕机、新 worker 加入、connector 配置改了 task 数）会触发再平衡——leader worker 重新计算"哪个 task 该在哪个 worker 上"。关键在于怎么过渡到新方案：

• 旧的 eager 策略（2.3 之前）：再平衡一开始，所有 worker 立即放下手上全部 task（revoke everything），等 leader 算出新分配后再统一领回。整个集群在这期间停止搬运——一个 worker 的加入会让所有 connector 短暂停摆，这就是 stop-the-world。
• incremental cooperative（2.3 起，KIP-415）：leader 先算出新旧分配的差集——只有真正需要换主的 task 才被收回，其余 task 完全不受影响、继续搬。收回的 task 在下一轮再分配给目标 worker。整个过程"增量"完成，集群绝大部分吞吐不中断。

对一个跑着几十个 connector 的集群，差别是质变：扩容时加一台 worker，eager 会让全部 connector 抖一下，cooperative 只动那几个被迁移的 task。

备选方案对比

代价与失效模式

cooperative 不是没有锋利的边。worker 数在再平衡过程中继续变动会引入分配倾斜（KAFKA-12495）：滚动重启时若 worker 一个接一个地进出，多轮增量再平衡可能收敛到一个 task 分布不均的中间态，部分 worker 过载、部分空闲。它不影响正确性，但会让吞吐不均，需要等集群稳定后再触发一次再平衡才会重新摊平。

与下一个机制的关系：再平衡决定 task 落在哪个 worker 上跑——而 task 真正搬运时，要把结构化记录变成 Kafka 里的字节。这一步的机制独立于 connector，发生在字节层：序列化机制。

2.3序列化机制：converter 在字节层定格式，与 connector 解耦

序列化由 converter 在字节层完成、与 connector 完全解耦；走 Schema Registry 的格式在字节前缀里写 magic byte + schema id，而 JsonConverter 与 JsonSchemaConverter 是两种不可互换的 wire format。

运行方式

01 章（§1.4）讲了 converter 与 connector 解耦"是什么"。这里看字节层"怎么工作"。一条记录在 Connect 内部是带 schema 的结构化对象（SchemaAndValue）；converter 负责把它变成 Kafka 里的字节，反向亦然。关键在于不同 converter 产出的字节布局完全不同：

• Schema Registry 系（Avro / Protobuf / JSON Schema converter）：字节不是裸数据，而是带前缀的 wire format——第 1 个字节是 magic byte（值 0x00），紧跟 4 字节的 schema id，之后才是序列化后的 payload。消费端读到 schema id，去 Schema Registry 拉对应 schema 再解码。schema 本身不随每条消息走，只走一个 id——这是它比内嵌 schema 省体积的原因。
• JsonConverter（org.apache.kafka.connect.json）：产出纯 JSON 字节，没有 magic byte。schemas.enable=true 时每条消息内嵌一份 {"schema":...,"payload":...}，false 时只发裸 JSON。
• JsonSchemaConverter（io.confluent.connect.json）：名字像 JSON，但它走 Schema Registry、字节带 magic byte + schema id。与 JsonConverter 的 wire format 完全不同，不可互换——这是最容易混淆、也最容易出事的一对。

解耦体现在：同一个 connector，换 converter 就换了线上字节格式，connector 代码与配置一行不动（图 1.2 已示）。这把"序列化"从每个 connector 各实现一遍，变成一个可插拔的横切层。

备选方案对比

代价与失效模式

解耦的代价是sink 端的 converter 必须与 topic 里实际的字节格式严格一致，否则就是反序列化阶段的硬失败。最典型的失效是 sink converter 与线上格式不符引发的反序列化风暴：上游 producer 写 Avro（字节以 0x00 开头），sink 配了 JsonConverter，每一条记录在反序列化时都报 Unknown magic byte!。由于 errors.tolerance 默认 none（§1.7），第一条就让整个 task FAILED。converter 不会"猜"格式、不会默默降级——配错就是失败。

{
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://schema-registry:8081",
  "key.converter": "org.apache.kafka.connect.storage.StringConverter"
}

与下一个机制的关系：converter 把记录序列化进 Kafka 后，task 还得记住"搬到哪了"以便重启续传。而 source 与 sink 的偏移存法和恢复路径完全不同——这是偏移管理与恢复。

2.4偏移管理与恢复：source 存 topic、sink 用消费组，两套各自续传

source connector 把自定义偏移存进 connect-offsets、按 connector 名索引，sink connector 用消费组（组名 connect-<name>）的位点；重启时各自从上次提交处续，REST /offsets（KIP-875）可读写两者。

运行方式

01 章（§1.6）建立了"两套偏移"这件事。这里看恢复路径怎么走。两套机制的差别源于一个事实：外部系统没有 Kafka 式的 offset 概念。

• source 恢复：外部系统（DB、文件）的"位点"由 connector 自己定义——"读到 id=10042""文件读到 byte 4096"。框架把这组 {sourcePartition → sourceOffset} 键值对周期性提交进 connect-offsets topic，key 含 connector 名。worker 重启或 task 被再平衡到别处，新实例先按自己的 connector 名去 connect-offsets 读回最近一次提交的偏移，再从那里继续向外部系统要数据。
• sink 恢复：sink 就是一个 Kafka 消费者，用普通消费组位点，组名固定为 connect-<connector 名>，位点存在 __consumer_offsets。重启即普通消费者重启——从消费组上次提交的位点接着消费，与任何 Kafka 消费者完全一样。

"按 connector 名索引"在 source 端埋了一个直接后果：给 source connector 改名 = 新名字在 connect-offsets 里查不到任何偏移 = 框架认为它从未搬过 = 从头重灌整个外部源。sink 端同理——改名意味着新消费组，从头消费。KIP-875（3.6）加了 REST GET/PATCH/DELETE /connectors/{name}/offsets，可以查看和手动调整两类偏移，但前提仍是名字稳定。

备选方案对比

代价与失效模式

分两套的代价集中在一个失效模式：改 source connector 名 = 丢偏移 = 重灌。想"重建"一个 source connector 而顺手把名字从 orders-jdbc-source 改成 orders-jdbc-source-v2，新名字在 connect-offsets 里对应一组空偏移，connector 会把整张表从头再搬一遍——对大表是一次代价高昂的全量重灌，还可能给下游灌入重复数据。

# 查看当前偏移（source 返回自定义位点；sink 返回消费组位点）
curl http://worker:8083/connectors/orders-jdbc-source/offsets

# 要重置时：先 stop，再 DELETE 偏移，而不是改名
curl -X PUT    http://worker:8083/connectors/orders-jdbc-source/stop
curl -X DELETE http://worker:8083/connectors/orders-jdbc-source/offsets

与下一个机制的关系：默认的偏移提交是"先写记录、再提交偏移"，崩溃时这两步之间的窗口会导致重发——即"至少一次"。要做到"精确一次",必须把这两步绑成原子操作。这是 EOS source。

2.5EOS source：事务双写 + 代际 fencing，worker 级开关

EOS source（KIP-618）把"写记录"与"提交偏移"放进同一个 Kafka 事务原子完成，再用代际 fencing 拒绝僵尸 task 的写入；它是 worker 级开关 exactly.once.source.support，仅 distributed 可用、须全集群一致。

运行方式

默认 source 是"至少一次"：task 先把一批记录写进目标 topic，再把对应偏移提交进 connect-offsets。如果崩溃发生在"记录已写、偏移未提交"这个窗口，重启后 task 从旧偏移续传，已写过的那批记录被重发。EOS source 用两个 Kafka 原语消除这个窗口：

• 事务双写：task 把"一批数据记录"和"对应的偏移更新（写进 connect-offsets）"放进同一个 Kafka 事务，原子提交。要么记录与偏移一起生效，要么都不生效——"记录写了、偏移没写"的窗口不复存在。
• 代际 fencing（僵尸隔离）：再平衡后旧 task 实例可能还"活着"想继续写（僵尸）。框架给每代 task 分配递增的事务标识，broker 只接受最新代的事务、拒绝旧代的写入。同一份工作因此只有最新一代能提交，僵尸写不进去。

开启是两阶段滚动：先把所有 worker 的 exactly.once.source.support 设为 preparing 滚动重启一轮，再设为 enabled 滚动重启一轮。两阶段是为了让集群在升级过程中始终保持一致——不能一半 worker 开、一半没开。connector 侧再声明 exactly-once.support=required 表示它要求这个保证。

备选方案对比

代价与失效模式

EOS source 的代价主要是约束而非性能：仅 distributed 模式可用（standalone 没有这套机制）、是 worker 级开关、必须全集群一致（不能 per-connector 单独开）、开启要两阶段滚动。最常见的失效是认知性的：以为能"只给账务这一个 connector 开 EOS"——做不到，要么整个 distributed 集群一起开，要么都不开。此外事务提交本身有开销，会略增延迟。

# 两阶段：先全员 preparing 滚动重启，再全员 enabled 滚动重启
exactly.once.source.support=enabled

2.6跨机制综合：一条记录如何穿过整个集群

五个机制不是孤立的——一次真实的搬运同时用到协调、再平衡、序列化、偏移恢复。把它们串成一个场景，看它们如何协同。

场景：一个 distributed 集群（worker 1/2/3，exactly.once.source.support 未开）跑着一个 JDBC source connector，从 orders 表增量读、写进 pg-orders topic。某时刻 worker 2 宕机。跟着一条记录走一遍：

1. task 分配（协调 §2.1 + 再平衡 §2.2）：connector 启动时，leader worker 按 tasks.max 与可切分单元算出 task，均摊到三个 worker——假设 orders 表的 task d 落在 worker 2 上。这份分配方案写进 connect-configs。
2. 读取 + 序列化（序列化 §2.3）：worker 2 上的 task d 从 orders 读到一行（id=10042），在 Connect 内部是结构化对象；value.converter（设为 Avro）把它序列化成带 magic byte + schema id 的字节，写进 pg-orders。
3. 提交偏移（偏移 §2.4）：框架周期性把 task d 的位点 {table:orders → incrementing:10042} 提交进 connect-offsets，key 含 connector 名。
4. worker 2 宕机 → 再平衡（§2.2）：消费组协议检测到 worker 2 离开，触发 incremental cooperative 再平衡。worker 1/3 上无关的 task 继续搬，只有 task d 被重新分配——假设迁到 worker 3。
5. 从偏移续传（§2.4 + §2.1）：worker 3 上新起的 task d 实例，先按 connector 名去 connect-offsets 读回最近提交的偏移（incrementing:10042），从 id>10042 继续向 orders 要数据——不重读已搬过的行。整个恢复没碰任何外部协调器，全程只读写 Kafka 的 topic。

这一圈把控制面（§2.1 三个 topic）、再平衡（§2.2 只动 task d）、序列化（§2.3 converter 定字节）、偏移恢复（§2.4 按名读回）咬合在一起。注意一个边界：因为这个集群没开 EOS（§2.5），第 3 步与第 2 步之间存在窗口——若 worker 2 恰在"记录已写、偏移未提交"时宕机，task d 在 worker 3 上会从上一个已提交偏移续传，重发 10042 这批里已写的记录（至少一次）。要消除这点，才需要 §2.5 的事务双写。