tencent cloud

import ( 
  "git.code.oa.com/vicenteli/trpc-database/kafka"
)

func init() {
  // 重写默认的异步生产写数据错误回调
  kafka.AsyncProducerErrorCallback = func(err error, topic string, key, value []byte, headers []sarama.RecordHeader) {
    // do something if async producer occurred error.
  }

  // 重写默认的异步生产写数据成功回调
  kafka.AsyncProducerSuccCallback = func(topic string, key, value []byte, headers []sarama.RecordHeader) {
    // do something if async producer succeed.
  }
}

address: fake_address 

func main() {
  s := trpc.NewServer()
  // 使用自定义 addr，需在启动 server 前注入
  cfg := kafka.GetDefaultConfig()
  cfg.Brokers = []string{"127.0.0.1:9092"}
  cfg.Topics = []string{"test_topic"}
  kafka.RegisterAddrConfig("fake_address", cfg)
  kafka.RegisterKafkaConsumerService(s.Service("fake_address"), &Consumer{})

  s.Serve()
}

通过 kafka.GetRawSaramaContext 可以获取底层 sarama ConsumerGroupSession 和 ConsumerGroupClaim。但是此处暴露这两个接口只是方便用户做监控日志，应该只使用其读方法，调用任何写方法在这里都是未定义行为，可能造成未知结果。
// RawSaramaContext 存放 sarama ConsumerGroupSession 和 ConsumerGroupClaim
// 导出此结构体是为了方便用户实现监控，提供的内容仅用于读，调用任何写方法属于未定义行为
type RawSaramaContext struct {
    Session  sarama.ConsumerGroupSession
    Claim sarama.ConsumerGroupClaim
}

func (Consumer) Handle(ctx context.Context, msg *sarama.ConsumerMessage) error {
    if rawContext, ok := kafka.GetRawSaramaContext(ctx); ok {
        log.Infof("InitialOffset: %d", rawContext.Claim.InitialOffset())
    }
    // ...
    return nil
}