偏移量与提交

__consumer_offsets 内部 Topic、Offset 提交不当的后果（重复消费/消息丢失），以及 franz-go 中自动/标记/手动三档提交的实现方式（kafka-go 存量写法在表格与注记里保留）。

特殊 Topic：__consumer_offsets

Kafka 通过一个名为 __consumer_offsets 的内部 Topic 来记录每个消费者组在每个分区上的消费进度（Offset）。

__consumer_offsets 的主要作用是：当消费者组发生重平衡后，每个消费者可以从上次提交的 Offset 处继续消费，避免消息丢失或重复。

franz-go 的进度管理集中在客户端：组客户端自己记账，按提交策略把"每组每分区最新 Offset"写进这个内部 Topic——PollFetches 拿到消息不会自动动进度，什么时候提交由选项决定（下文三档）。kafka-go 存量对应 ReadMessage（读后即提）与 FetchMessage + CommitMessages（手动）。

它长什么样

“进度"本身就是消息：每次提交往该 Topic 写一条 Key = 组名+分区、Value = Offset+元数据 的记录，靠 compact 清理策略保证每组每分区永远只留最新一条（机制见 11 章）。由此推出几个默认参数与含义：

最后一行藏着一个经典事故：空跑一周的组回来接着消费时，Offset 已不存在，起点取决于客户端的"位点失效恢复策略”——Java 按 auto.offset.reset（默认 latest，等于把一周消息全跳了）；franz-go 默认 ConsumeStartOffset(AtStart) + ConsumeResetOffset(RewindOffset(1min))，新组/丢进度会从 Topic 头全量重放；kafka-go 的 StartOffset 零值同样等价 FirstOffset（也是全量重放）。灾备备用消费者、季节性批处理这类"休眠组"上线前，要么调大 offsets.retention.minutes，要么显式声明起点，别赌默认。

按时间/边界定位 Offset

Offset 是"从哪读"的坐标，运维和初始化都绕不开"按时间找 Offset"：

# CLI：查各分区某时刻对应的 Offset（-1 最新 / -2 最早）
.\bin\windows\kafka-get-offsets.bat --bootstrap-server localhost:9092 \
  --topic order-events --time 1762000000000

// 代码定位：非组模式（03 章的 scan 客户端）在订阅时就声明每个分区的起点
scan, _ := kgo.NewClient(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumePartitions(map[string]map[int32]kgo.Offset{
        "order-events": {
            0: kgo.NewOffset().AfterMilli(time.Now().Add(-time.Hour).UnixMilli()), // 按时间：约 1 小时前起
            1: kgo.NewOffset().AtEnd().Relative(-1000),                             // 按位点：倒数 1000 条起
            2: kgo.NewOffset().At(11223344),                                        // 精确 Offset
        },
    }),
)

// 新组起点同理可用时间：kgo.ConsumeStartOffset(kgo.NewOffset().AfterMilli(...))

franz 的 kgo.Offset 是"起点配方"而不是数字，At/AtStart/AtEnd/Relative/AfterMilli 五种写法最终都由客户端换算成分区里的具体位置——AfterMilli 底层就是 ListOffsets 按时间定位（等价 CLI 的 --time）。kafka-go 存量对应低层 conn.ReadOffsets() 拿首尾 + conn.SeekT/SeekL 移动游标，思路一样、API 更手动。

注意：按时间定位是"定位到第一条 ≥ 目标时间的消息"的近似——时间戳非单调的 Topic（CreateTime 且生产者乱序）会有偏差。精确回放优先用按业务时间过滤，Offset 只负责定位到大概位置。

组进度重置（运维操作）

改代码起点只能影响新组；存量组要"回到某个时刻重放/跳过积压"用 kafka-consumer-groups --reset-offsets（完整场景见 02 章）：

# 先 dry-run 看效果（不加 --execute 不会真的改）
.\bin\windows\kafka-consumer-groups.bat --bootstrap-server localhost:9092 \
  --group order-processors --topic order-events --reset-offsets --to-datetime 2026-10-13T00:00:00.000Z
# 确认输出无误后，同一命令追加 --execute

前提是组内没有活跃消费者（组必须为空，否则拒绝执行）——先停应用再重置再启动，顺序别反。程序化重置走 kadm：adm.CommitOffsets(ctx, group, offsets)（直接改写组进度）/ adm.DeleteOffsets(ctx, group, topics)（删进度，让组按起点策略重来）/ adm.DeleteRecords（给分区打删除下界，配合 11 章合规场景）。

Offset 提交不当的后果

提交的 Offset 小于实际消费到的位置 → 消息被重复消费

提交的 Offset 大于实际消费到的位置 → 中间的消息丢失

Offset 提交方式（franz-go 三档）

1. 自动提交 — 默认行为

什么都不配，组客户端就处于自动提交档：后台每 AutoCommitInterval（默认 5s）提交已消费到的位置。

// 03 章的 group 客户端保持默认即为自动提交档
group, _ := kgo.NewClient(
    /* ...SeedBrokers / ConsumerGroup / ConsumeTopics / 起点... */
    kgo.AutoCommitInterval(2 * time.Second), // 想更勤就调小；代价是提交请求变密
)

优点：省心，适合"重复消费可接受、丢一条不致命"的日志/指标类链路。 缺点：拿到消息即被视为"已消费"，处理失败或两次提交间隔内宕机，这段消息会被重复消费；反之它也不会帮你"失败重放"——语义是至多一次的加强版（约等于"至少一次但区间可能拉长"，取决于崩溃时机）。业务链路慎用。

2. 标记式提交 — 生产推荐折中档

AutoCommitMarks()：进度由你自己 MarkCommitRecords 打点，后台按周期批量提交打过的点——把"什么时候算消费成功"交给业务，把"什么时候发请求"交给框架。

group, _ := kgo.NewClient(
    /* ...组选项... */
    kgo.AutoCommitMarks(), // 关掉"读后即提"，只提交 Mark 过的
)

fs := group.PollRecords(ctx, 500)
var done []*kgo.Record
fs.EachRecord(func(r *kgo.Record) {
    if err := handle(r); err != nil {
        return // 失败的不打点 → 下次重平衡/重启后重放
    }
    done = append(done, r)
})
group.MarkCommitRecords(done...) // 记账；提交动作由后台周期完成

宕机时最多丢"最后一个周期内已打点未提交"的窗口，重复量可控；每条都打点也不会每条都发网络请求。绝大多数业务组用这一档就够。

3. 全手动提交 — 精确控制

DisableAutoCommit() + 处理成功后显式提交：

group, _ := kgo.NewClient(/* ...组选项... */, kgo.DisableAutoCommit())

for {
    fs := group.PollRecords(ctx, 500)
    if err := fs.Err0(); errors.Is(err, context.Canceled) || errors.Is(err, kgo.ErrClientClosed) {
        return // 收信号退出：在途批次此刻未提交，重启后重放，靠幂等兜底
    }

    var done []*kgo.Record
    fs.EachRecord(func(r *kgo.Record) {
        if err := handle(r); err != nil {
            log.Printf("处理失败 topic=%s partition=%d offset=%d: %v",
                r.Topic, r.Partition, r.Offset, err)
            return // 失败不提交 → 消息会被重新消费
        }
        done = append(done, r)
    })

    if len(done) > 0 {
        // 同步提交一批记录：每分区取最大 Offset（累积确认语义）
        if err := group.CommitRecords(ctx, done...); err != nil {
            log.Printf("提交失败: %v", err)
        }
    }
}

三个变体按场景选：

• CommitRecords(ctx, rs...)：同步、按记录提交，同批取每分区最大 Offset——所以给半批记录就等于把该分区"已消费"的边界推进到批尾；要精确到"跳过中间某条"就单独提交那条。
• CommitOffsetsSync(ctx, map[string]map[int32]kgo.EpochOffset{...})：同步、按显式位点提交，适合"从快照恢复进度"的场景。
• CommitOffsets(ctx, offsets, func(error))：异步提交（带回调）——高吞吐链路连续提交不想阻塞消费循环时用；注意它可能在组关闭时丢回调，退出前调 CommitUncommittedOffsets(ctx) 或在 Close() 前留同步收尾。

提交规则：提交某个 Offset 时，该分区所有更小的 Offset 也会被一并标记为已消费——与 TCP 的累积确认机制类似。franz 的 MarkedOffsets()/CommittedOffsets() 可以随时对账"本地打了什么点 / 服务端记了什么位"。

Java vs franz-go Offset 提交对照

kafka-go 存量速记：ReadMessage≈自动提交、FetchMessage+CommitMessages≈手动、CommitInterval≈自动周期；没有异步提交、没有标记式，进度控制粒度到"批"为止。迁移到 franz 时对应关系直接查上表。