tencent cloud

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

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

  // Rewrite the default callback for successes of asynchronous production writing data.
  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()
  // Use custom addr, which should be injected before the server starts
  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()
}

By using kafka.GetRawSaramaContext, you can access the underlying Sarama ConsumerGroupSession and ConsumerGroupClaim. However, exposing these two APIs here is just to facilitate users to monitor logs. Only their read methods should be used, as calling any write methods here is undefined behavior, which may cause unknown results.
// RawSaramaContext stores Sarama ConsumerGroupSession and ConsumerGroupClaim
// Export this structure to facilitate user monitoring. The content provided is for read-only purposes, as calling any write methods is 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
}

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

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

  // Rewrite the default callback for successes of asynchronous production writing data.
  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()
  // Use custom addr, which should be injected before the server is started.
  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()
}

By using kafka.GetRawSaramaContext, you can access the underlying Sarama ConsumerGroupSession and ConsumerGroupClaim. However, exposing these two APIs here is just to facilitate users to monitor logs. Only their read methods should be used, as calling any write methods here is undefined behavior, which may cause unknown results.
// RawSaramaContext stores sarama ConsumerGroupSession and ConsumerGroupClaim
// Export this structure to facilitate user monitoring. The content provided is for read-only purposes, as calling any write methods is 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
}