可靠性与幂等消费

// 发送一条订单事件；失败分类处理，最终兜底进本地消息表
func publish(ctx context.Context, cl *kgo.Client, orderID string, payload []byte) error {
    rec := &kgo.Record{
        Key:   []byte(orderID),
        Value: payload,
        // Topic 由 cl 的 DefaultProduceTopic("order-events") 兜底（03 章）
    }
    withMsgID(rec, snowflake()) // 全局 msgID 进 Header，去重键设计见下文

    ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    err := cl.ProduceSync(ctx, rec).FirstErr()
    if err == nil {
        return nil
    }

    var ke *kerr.Error
    switch {
    // Leader 切换/迁移中：franz 内部已重试过，走到这里说明超时太久，直接交给兜底
    case errors.As(err, &ke) && (ke.Code == kerr.LeaderNotAvailable.Code ||
        ke.Code == kerr.NotLeaderForPartition.Code):
    // ISR 不足（Broker 端 min.insync.replicas 挡住写入），恢复副本后才有救
    case errors.As(err, &ke) && (ke.Code == kerr.NotEnoughReplicas.Code ||
        ke.Code == kerr.NotEnoughReplicasAfterAppend.Code):
    // 消息超限额（MessageTooLarge），重试一万次也不会成功
    case errors.As(err, &ke) && ke.Code == kerr.MessageTooLarge.Code:
        log.Printf("消息过大被拒: orderID=%s", orderID)
    }

    // 兜底：写本地消息表，由后台补偿任务重投（见下方"本地消息表"）
    return saveToOutbox(rec, err)
}

// outbox 表：id, topic, part_key, payload, err_msg, status(pending/sent), created_at
// 与订单写入同一事务提交，保证"业务成功则消息必发"
func saveToOutbox(rec *kgo.Record, cause error) error {
    topic := rec.Topic
    if topic == "" {
        topic = "order-events" // DefaultProduceTopic 兜底值，落表前补全
    }
    return db.Exec(
        `INSERT INTO kafka_outbox (topic, part_key, payload, err_msg, status, created_at)
         VALUES (?, ?, ?, ?, 'pending', NOW())`,
        topic, string(rec.Key), rec.Value, cause.Error(),
    ).Error
}

# 1. 建 Topic 时副本至少 3（单机开发无所谓，集群必须）
.\bin\windows\kafka-topics.bat --create --topic order-events \
  --bootstrap-server localhost:9092 --partitions 6 --replication-factor 3

# 2. ISR 中至少 2 个副本落盘才算写成功（3 副本容忍挂 2 个而不丢？不是——
#    是"挂 1 个还能写、挂 2 个拒绝写入"，用可用性换持久性）
.\bin\windows\kafka-topics.bat --alter --topic order-events \
  --bootstrap-server localhost:9092 --config min.insync.replicas=2

# 3. 禁止脏选举：宁可分区不可用，也不能让落后副本当 Leader 后抹掉数据
.\bin\windows\kafka-configs.bat --bootstrap-server localhost:9092 \
  --entity-type brokers --entity-default --alter \
  --add-config unclean.leader.election.enable=false

// 03 章的全手动循环（group 已配 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 {
            // 不提交：本条以及后续都会被重放；重试上限与死信处理见 10
            log.Printf("处理失败 offset=%d: %v", r.Offset, err)
            return
        }
        done = append(done, r)
    })
    if len(done) > 0 {
        if err := group.CommitRecords(ctx, done...); err != nil {
            // 提交失败也要处理，否则这批会重复消费——好在幂等兜着
            log.Printf("提交失败: %v", err)
        }
    }
}

// 生产端：塞一个全局 msgID 进 Header（franz 的 Header 是 []kgo.RecordHeader{Key string; Value []byte}）
func withMsgID(rec *kgo.Record, msgID string) *kgo.Record {
    rec.Headers = append(rec.Headers, kgo.RecordHeader{Key: "x-msg-id", Value: []byte(msgID)})
    return rec
}

// 消费端：从 Header 取
func msgIDOf(hs []kgo.RecordHeader) string {
    for _, h := range hs {
        if h.Key == "x-msg-id" {
            return string(h.Value)
        }
    }
    return "" // 没有则退回业务键
}

// 依赖 go-redis；key 带 Topic 维度避免串号
func once(ctx context.Context, rdb *redis.Client, r *kgo.Record) (bool, error) {
    key := fmt.Sprintf("kafka:dedup:%s:%s", r.Topic, msgIDOf(r.Headers))
    // TTL > 消息最大重放窗口（retention + 重试时长），但要短于业务允许的"同 ID 再来"间隔，
    // 正常一条业务消息的 ID 本来就是唯一的，TTL 3~7 天足够
    return rdb.SetNX(ctx, key, 1, 72*time.Hour).Result()
}

// 消费循环里（handle 之前判）：
ok, err := once(ctx, rdb, r)
if err != nil { /* Redis 故障降级：宁可重复处理，不要卡消费 */ }
if !ok {
    done = append(done, r) // 重复消息：视为已处理，正常提交跳过
    return
}

// INSERT IGNORE 或 ON DUPLICATE KEY，靠 (topic, msg_id) 唯一索引判重
res, err := db.ExecContext(ctx,
    `INSERT INTO consumed_msg (topic, msg_id, created_at) VALUES (?, ?, NOW())
     ON DUPLICATE KEY UPDATE msg_id = msg_id`, r.Topic, msgIDOf(r.Headers))
if err != nil { /* 回滚重试 */ }
if n, _ := res.RowsAffected(); n == 0 {
    // 影响行数 0 = 该消息已消费过，直接提交 Offset 跳过
}

// 事件携带"目标状态 + 前驱状态"，UPDATE 只在前驱匹配时生效
res, _ := db.ExecContext(ctx,
    `UPDATE orders SET status = 'paid'
     WHERE order_no = ? AND status = 'created'`, orderNo)
n, _ := res.RowsAffected()
if n == 0 {
    // 状态已推进过（重复消息），幂等地提交跳过
}

// franz-go：消费 order-raw → 转换 → 写 order-clean，全程精确一次
gts, err := kgo.NewGroupTransactSession(
    kgo.SeedBrokers("localhost:9092"),
    kgo.ConsumerGroup("clean-order"),
    kgo.ConsumeTopics("order-raw"),
    kgo.TransactionalID("order-cleaner-1"), // 每个实例唯一，重启复用同一 ID
)
defer gts.Close()

ctx := context.Background()
for {
    fetches := gts.PollRecords(ctx, 100)
    if err := fetches.Err0(); errors.Is(err, context.Canceled) || errors.Is(err, kgo.ErrClientClosed) {
        return
    }
    if err := gts.Begin(); err != nil {
        continue // 事务开启失败：franz-go 内部已恢复/重试
    }

    var recs []*kgo.Record
    fetches.EachRecord(func(r *kgo.Record) {
        recs = append(recs, &kgo.Record{
            Topic: "order-cleaner-input",
            Key:   r.Key,
            Value: transform(r.Value),
        })
    })

    // 生产 + Offset 提交都登记在本事务内
    errs := gts.ProduceSync(ctx, recs...).Errs()
    endAs := kgo.TryCommit
    if len(errs) > 0 {
        endAs = kgo.TryAbort // 生产失败：整笔事务回滚，下次 poll 重放原消息
    }
    committed, err := gts.End(ctx, endAs)
    if err != nil || !committed {
        // 事务未成功（fencing/超时等）：franz-go 会自动恢复事务状态，消息重新投递
    }
}