重试、死信与延迟消息

// 重试等级表：Topic 名 + 本级停留时长
var retryLevels = []struct {
    topic string
    hold  time.Duration
}{
    {"order-events-retry-10s", 10 * time.Second},
    {"order-events-retry-1m", time.Minute},
    {"order-events-retry-10m", 10 * time.Minute},
}

// bumpToRetry 把失败消息打入下一级重试 Topic，attempt +1。
// r 来自 group 消费者的 PollRecords：Key/Value/Headers 原样带着走，
// murmur2 分区器保证同 Key 在重试 Topic 里也进同一分区号。
func bumpToRetry(ctx context.Context, cl *kgo.Client, r *kgo.Record, attempt int) error {
    if attempt >= len(retryLevels) {
        // 阶梯用尽 → 死信（带失败原因，方便排查）
        r.Headers = setHeader(r.Headers, "x-fail-topic", r.Topic)
        r.Headers = setHeader(r.Headers, "x-fail-attempt", strconv.Itoa(attempt))
        r.Topic = "order-events-dlt"
        return cl.ProduceSync(ctx, r).FirstErr()
    }
    lvl := retryLevels[attempt]
    r.Topic = lvl.topic
    r.Headers = setHeader(r.Headers, "x-attempt", strconv.Itoa(attempt+1))
    r.Headers = setHeader(r.Headers, "x-ready-at", strconv.FormatInt(
        time.Now().Add(lvl.hold).UnixMilli(), 10))
    return cl.ProduceSync(ctx, r).FirstErr()
}

// setHeader 覆盖式写入 Header（同 key 只保留一个值）；[]kgo.RecordHeader 与 []kafka.Header 同构
func setHeader(hs []kgo.RecordHeader, key, val string) []kgo.RecordHeader {
    for i := range hs {
        if hs[i].Key == key {
            hs[i].Value = []byte(val)
            return hs
        }
    }
    return append(hs, kgo.RecordHeader{Key: key, Value: []byte(val)})
}

// 消费 order-events-retry-10s 的循环（重试组客户端按 03 章 group 的建法换一个
// ConsumerGroup/ConsumeTopics 即可）。"延迟"的实现：
// 拉取后若未到 x-ready-at，就地等一等再处理（单分区流量小，sleep 不伤大雅）；
// 更讲究的做法是把"未到期消息重新投回本 Topic 并更新 ready-at"，避免占住消费循环。
func consumeRetry(ctx context.Context, retryGroup *kgo.Client, cl *kgo.Client) {
    for {
        fs := retryGroup.PollRecords(ctx, 200)
        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 ready := readyAtOf(r.Headers); ready > 0 {
                if wait := time.Until(time.UnixMilli(ready)); wait > 0 {
                    select {
                    case <-time.After(wait):
                    case <-ctx.Done():
                        return // 未提交，关机会重放，靠幂等兜底
                    }
                }
            }

            attempt := attemptOf(r.Headers) // 从 x-attempt Header 解析，缺省 0
            if err := handle(r); err != nil {
                if err2 := bumpToRetry(ctx, cl, r, attempt); err2 != nil {
                    log.Printf("转入重试通道失败: %v", err2) // 不提交，留在本 Topic 重放
                    return
                }
                done = append(done, r) // 已成功转走，主流程推进
                return
            }
            done = append(done, r)
        })
        if len(done) > 0 {
            retryGroup.CommitRecords(ctx, done...)
        }
    }
}

func attemptOf(hs []kgo.RecordHeader) int {
    for _, h := range hs {
        if h.Key == "x-attempt" {
            n, _ := strconv.Atoi(string(h.Value))
            return n
        }
    }
    return 0
}

// readyAtOf 与 attemptOf 同构：从 x-ready-at Header 解出毫秒时间戳，缺省 0（立即可处理）
func readyAtOf(hs []kgo.RecordHeader) int64 {
    for _, h := range hs {
        if h.Key == "x-ready-at" {
            ms, _ := strconv.ParseInt(string(h.Value), 10, 64)
            return ms
        }
    }
    return 0
}

# 主/重试/DLT 一起建；DLT 不设置过期，进一条留一条
.\bin\windows\kafka-topics.bat --create --topic order-events-dlt \
  --bootstrap-server localhost:9092 --partitions 1 --replication-factor 3
.\bin\windows\kafka-configs.bat --bootstrap-server localhost:9092 \
  --entity-type topics --entity-name order-events-dlt --alter \
  --add-config retention.ms=-1

# 导出全部死信 → 逐行回投主 Topic（消息体不变；attempt 头在导出时丢失，回放后重新计数）
.\bin\windows\kafka-console-consumer.bat --topic order-events-dlt --from-beginning \
  --bootstrap-server localhost:9092 --property print.key=true \
  --property key.separator="|" --max-messages 100000 > dlt_dump.txt

# 按 key|value 格式回投（控制台生产者默认 tab 分隔，用 property 对齐）
.\bin\windows\kafka-console-producer.bat --topic order-events \
  --bootstrap-server localhost:9092 --property parse.key=true \
  --property key.separator="|" < dlt_dump.txt

// 与 ①② 共用的 retryLevels 类似，建一组 delay-<级别> Topic
// order:cancel-30m 由专门的消费者订阅，到期后投回业务 Topic
func sendDelayed(ctx context.Context, cl *kgo.Client, levelTopic string, r *kgo.Record, fireAt time.Time) error {
    r.Topic = levelTopic // 同样靠 Record.Topic 覆盖 DefaultProduceTopic
    r.Headers = setHeader(r.Headers, "x-ready-at", strconv.FormatInt(fireAt.UnixMilli(), 10))
    return cl.ProduceSync(ctx, r).FirstErr()
}