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

func init() {
  // Override the default error callback for asynchronous message production
  kafka.AsyncProducerErrorCallback = func(err error, topic string, key, value []byte, headers []sarama.RecordHeader) {
    // do something if async producer occurred error.
  }

  // Override the default success callback for asynchronous message production
  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()
  // To use a custom addr, it must be injected before starting the 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()
}

Through kafka.GetRawSaramaContext, you can obtain the underlying sarama ConsumerGroupSession and ConsumerGroupClaim. However, exposing these two interfaces here is solely for facilitating user monitoring and logging. Only their read methods should be used, as calling any write methods constitutes undefined behavior that may lead to unknown results.
// RawSaramaContext holds the sarama ConsumerGroupSession and ConsumerGroupClaim
// Exporting this structure is for the convenience of users to implement monitoring. The provided content is for read-only purposes. Calling any write methods constitutes undefined behavior.
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
}