Kafka 消费位点、并行处理与重平衡

固定 Kafka 3.7.1 Java consumer经典组协议。先学生产与复制，再把本页的提交规则用于业务幂等。新协议实现不能直接套入经典JoinGroup/SyncGroup调用链。

三种进度和一个空洞

消费端至少区分poll position、业务完成边界、committed offset。poll返回100到109后position可能到110；业务只完成100到102；提交103表示下次从103开始，不是“第103条已完成”。fetch进度是客户端拿到了什么，业务进度是效果落到了哪里，提交进度是故障恢复从哪里开始，三者不同。

自动提交不知道外部线程池何时完成数据库事务。把记录派给worker后继续poll，自动提交可能越过尚未完成任务。手动提交也不天然安全：101、102完成而100未完成时提交103，重启会遗漏100。可靠并行消费必须为每分区维护有序交付队列，只推进连续完成的交付前缀。

连续指应用已收到的记录，不指每个整数offset。事务控制批次、过滤和日志压缩可能产生间隙。实际交付[100,103,108]，前两条完成即可提交104；不必等待永远不会交付的101、102，也不能跨过未完成108提交109。这个细节决定实现应维护交付顺序，而不是从起点机械加一。

外部数据库与__consumer_offsets是两个持久边界。先业务后提交会重放，先提交后业务会遗漏。常见设计接受重放并在数据库同事务保存Inbox与业务效果。若读Kafka写Kafka，可用事务输出与offset；若把业务位点保存在自有数据库，可原子保存效果与位点并在分配时seek，但须自行处理所有权、恢复及监控，不能同时把Kafka提交和数据库提交都当唯一真相。

稀疏交付的手算轨迹

起始安全水位100；已交付100、103、108，但没有交付101/102/104–107。完成顺序108→100→103：先完成108只能记完成，不能跨100；完成100后提交101是保守且安全的；完成103后队头103与已完成108一同移出，可提交109。整数间隙无需“补完成通知”，但每个实际交付的空洞都必须等待。

这与“前两条100/103完成，108仍未完成时可提交104”不冲突。模型选择最后已完成offset+1作为保守边界；也可在协议允许时选择下一个未完成的已交付offset，只要不跨过未完成记录。不能把未交付的整数误作永远卡住的任务。

提交源码：持久化前先检查组身份

客户端commitSync → ConsumerCoordinator.commitOffsetsSync向组coordinator发OffsetCommit；broker经典入口为GroupCoordinator.handleCommitOffsets，验证组状态、member、generation和静态身份，再进入位点存储。此函数名是handleCommitOffsets，不是handleOffsetCommit。来源：ConsumerCoordinator、GroupCoordinator。

以下为教学改写，省略空组、事务offset等特殊分支：

教学伪代码

validate coordinator ownership and group state
validate memberId / groupInstanceId
if generationId != group.generationId:
    return ILLEGAL_GENERATION
persist offsets; respond

generation阻止旧代成员提交位点，但不会撤销已交给数据库的SQL。旧worker在撤销后恢复，新owner也从旧位点重放，两者可能同时做业务。“组内一个分区只有一个合法owner”不等于“外部数据库永远只有一个在途执行者”。资源端幂等、版本条件和租约fencing仍然必要。

commitSync等待结果并受超时与错误类型约束，catch异常后不能宣称已提交。commitAsync降低等待，但失败回调不能无条件重发旧offset：提交103失败、后续110成功后再发103会倒退恢复点，造成额外重放。让poll线程集中维护单调提交边界和请求序号，或在安全停机边界同步提交。倒退影响恢复位置，不等于消息被broker删除。

提交确认丢失也有UNKNOWN：协调器可能已保存位点。重试同一安全offset不会创造业务遗漏，但重发过期offset会扩大重复窗口。故障处理应把“安全可提交边界”和“最近已确认边界”分别存于内存，不因一次超时把安全边界清零，也不把所有已拉取记录视为安全。

原码：位点落盘之前先验证组身份

原码连续摘录（仅统一公共缩进，截取方法局部）：GroupCoordinator.scala，3.7.1，L1047–L1070。

原码

private def doCommitOffsets(group: GroupMetadata,
                            memberId: String,
                            groupInstanceId: Option[String],
                            generationId: Int,
                            offsetMetadata: immutable.Map[TopicIdPartition, OffsetAndMetadata],
                            responseCallback: immutable.Map[TopicIdPartition, Errors] => Unit,
                            requestLocal: RequestLocal): Unit = {
  group.inLock {
    val validationErrorOpt = validateOffsetCommit(
      group,
      generationId,
      memberId,
      groupInstanceId,
      isTransactional = false
    )

    val offsetTopicPartition = new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, partitionFor(group.groupId))

    if (validationErrorOpt.isDefined) {
      responseCallback(offsetMetadata.map { case (k, _) => k -> validationErrorOpt.get })
    } else {
      group.currentState match {
        case Empty =>
          groupManager.storeOffsets(group, memberId, offsetTopicPartition, offsetMetadata, responseCallback, verificationGuard = None)

整个路径在group.inLock内先validateOffsetCommit，再依据组状态走storeOffsets或返回错误。它校验成员、generation等信息，不检查你的数据库副作用是否完成。合法提交100也不证明99之前所有并行业务已成功，连续完成水位必须由消费者自身维护。

原码：当前成员合法仍须匹配 generation

固定 apache/kafka / 3.7.1，GroupCoordinator.scala L1023–L1035。以下连续原码只统一公共缩进，没有改写分支。

原码

    } else if (generationId >= 0 || memberId != JoinGroupRequest.UNKNOWN_MEMBER_ID || groupInstanceId.isDefined) {
      validateCurrentMember(
        group,
        memberId,
        groupInstanceId,
        operation = if (isTransactional) "txn-offset-commit" else "offset-commit"
      ).orElse {
        if (generationId != group.generationId) {
          Some(Errors.ILLEGAL_GENERATION)
        } else {
          None
        }
      }

validateCurrentMember 检查成员与静态身份，随后 generation 不匹配返回 ILLEGAL_GENERATION。它不调用你的worker、不看数据库流水，无法替外部副作用做所有权裁决。旧代的提交被拒绝，不代表旧代已经发出的SQL被取消。

Join、Sync、Heartbeat 到 cooperative

成员向coordinator JoinGroup，组leader执行assignor，SyncGroup下发分配，之后heartbeat维持成员状态。coordinator是broker角色，组leader是客户端分配角色，不能混淆。成员集合、订阅、coordinator故障或处理停滞均可能触发入组与分配流程。

AbstractCoordinator驱动入组与heartbeat；ConsumerCoordinator.onJoinPrepare/onJoinComplete处理提交准备、用户rebalance回调及assignment。eager协议通常先撤销全部分区再重新分配；CooperativeStickyAssignor根据ownedPartitions识别迁移，先撤销待转移分区，再在后续轮次授予新owner，减少不必要停顿。它不允许新旧成员同时合法持有同分区，也不等于没有rebalance。来源：AbstractCoordinator、CooperativeStickyAssignor。

cooperative回调应只处理实际撤销列表，不能每次全清内存并丢掉保留分区的进度。策略升级需让组成员协商到兼容协议，滚动迁移时不能假定混合配置必然直接得到cooperative行为。静态成员group.instance.id减少短暂重启的身份变化，重复身份有fencing风险；它不支持无限失联，更不保证业务恰好执行一次。

session.timeout关注heartbeat可达性，max.poll.interval关注应用两次poll间隔。后台heartbeat正常仍可能前台业务卡死。源码heartbeat线程明确检查pollTimeoutExpired，因此只增加session.timeout不能修复单批处理太慢。静态成员超过poll间隔后的接管时机还受session处理影响，不能承诺到点立即迁移。来源：3.7 consumer配置、上述AbstractCoordinator。

有界并行、撤销和优雅停机

大多数KafkaConsumer操作不具备线程安全性。一个poll线程负责poll、assignment、pause/resume、commit；worker仅处理不可变记录并回报完成。每分区维护交付队列、完成集合和assignment epoch。重新分配时更换本地epoch，旧worker完成通知不能推进新assignment位点；但旧worker的数据库写入仍需幂等保护。跨线程唤醒采用文档支持的wakeup，不把consumer对象随意共享给worker。

poll: append (offset, assignmentEpoch) to delivery queue
worker: commit business transaction; report completion
poll thread: discard stale completion epochs
while head is complete:
    safeOffset = popHead.offset + 1
commit safeOffset only for currently owned partitions

这是教学算法，工程实现还要处理失败任务、上限、关闭超时。到高水位pause对应分区并继续poll，低水位resume；pause不等于离组，也不会自动取消已交付任务。无限队列把下游慢转成内存和更长恢复时间；每次poll后无界等待worker又可能超过poll间隔。

正常revoke先停该分区派发，在预算内drain，再提交连续完成前缀；未完成部分留待重放。onPartitionsLost表示所有权可能已丢失，提交可能失败，不能当正常revoke。强制终止时依赖稳定业务ID恢复；优雅关闭则按停止入口、有限drain、提交安全边界、close推进。容器宽限必须覆盖摘流、worker排空和consumer关闭，给超时留下缓冲。

已有页面的发布协议补充在此落到分区交接：活性探针失败会触发重启，就绪失败停止接新流量，二者不能都绑定瞬时下游故障。过载时活性探针不断失败会形成重启风暴，offset反复重放进一步压垮数据库。新旧消费者共存时消息schema必须兼容，不能先发布要求新字段必填的消费者，再等待生产者升级。

撤销到新分配的恢复时间线

本地 epoch 不是 broker generation 的替代物：前者隔离应用完成通知，后者约束协调器提交。撤销前停止派发并有限排空；所有权丢失后不再尝试把旧任务完成当作当前分区可提交。新owner从已提交安全水位重放103/108，数据库仍可能已经由旧worker写过，必须回查业务幂等的原子边界。

带输入的验证

下载完整模型，独立运行时去掉命令中的目录前缀；仓库根运行：

实验代码

python3 labs/middleware-senior/reliability_lab.py offsets --json
python3 labs/middleware-senior/reliability_lab.py offsets --case stale --json
python3 labs/middleware-senior/reliability_lab.py offsets --case handoff --json

normal交付[100,103,108]、完成108/100/103，公开trace安全水位为100/101/109。脚本实际执行故意错误的max(completed)+1策略，首个反例为unsafe_offset109、skipped=[100,103]。stale把完成通知保留旧epoch，当前epoch变2，三次安全水位均100。

handoff先完成100→水位101；撤销后旧103通知记not-owner，不推进；epoch2的新owner重放[103,108]，旧epoch1的108通知记old-assignment。新108先完成仍101，随后103完成一次推进至109，重复108通知保持109。输出仅表示poll线程的安全可提交边界，不冒充远端已确认提交，也不声称旧worker业务从未执行。本轮已运行三入口及对应断言，没有启动真实消费组。

真实实验准备3.7.1三分区topic和两个同组console consumer，再运行：

实验代码

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group reliability-g1

记录owner、committed、lag，退出一实例后重新describe，预期分区转移。要验证业务窗口，需Java consumer关闭auto commit，在数据库event_id唯一约束下写效果；业务事务提交后、commitSync前kill，再启动，预期重放但余额只变一次。console consumer只能观察组转移，不能证明自定义worker正确。

故障变式将单批耗时设为超过max.poll.interval，再观察rejoin与提交错误；对照减少max.poll.records和有界worker。处理中位数5ms、p99为200ms时，不能拿中位数乘批大小预测最坏批耗时，应量取整个batch分布并为GC、下游慢留预算。本轮未启动多consumer集群，具体重平衡时间与日志不作为实测结论。

架构师进阶：扩容与顺序迁移

12分区、24同组消费者不会得到24份活跃分区owner；多实例可作冗余但不能突破分区粒度。增加分区可能改变默认key映射，旧记录与新记录分落不同分区；严格聚合顺序需要旧流排空屏障、固定路由迁移或业务版本约束。worker并行可大于分区数，但应用承担乱序、交付空洞和撤销管理成本。

纯Kafka变为数据库加支付后，Kafka事务无法覆盖支付。分区串行易解释但受慢消息阻塞；按key串行提高并行度但需调度和迁移协议；无序并行加版本条件提升吞吐但冲突需重试/隔离。验收指标应包括最老未完成年龄、撤销耗时、提交失败、幂等冲突次数与恢复时间。若消费者lag很低但最老任务已卡一小时，用户风险仍可能很高。

闭卷验收

本页 7 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。