Golang中如何正确的使用sarama包操作Kafka？

type msgConsumerGroup struct{}

func (msgConsumerGroup) Setup(_ sarama.ConsumerGroupSession) error   { return nil }

func (msgConsumerGroup) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

func (h msgConsumerGroup) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {

   for msg := range claim.Messages() {

      fmt.Printf("%s Message topic:%q partition:%d offset:%d  value:%s\n", h.name, msg.Topic, msg.Partition, msg.Offset, string(msg.Value))

      // 查mysql去重

      if check(msg) {

          // 插入mysql

          insertToMysql()

      }

      // 标记，sarama会自动进行提交，默认间隔1秒

      sess.MarkMessage(msg, "")

   }

   return nil

}

func main(){

    consumerConfig := sarama.NewConfig()

    consumerConfig.Version = sarama.V2_8_0_0 // specify appropriate version

    consumerConfig.Consumer.Return.Errors = false

    //consumerConfig.Consumer.Offsets.AutoCommit.Enable = true      // 禁用自动提交，改为手动

    //consumerConfig.Consumer.Offsets.AutoCommit.Interval = time.Second * 1 // 测试3秒自动提交

    consumerConfig.Consumer.Offsets.Initial = sarama.OffsetNewest

    cGroup, err := sarama.NewConsumerGroup([]string{"10.0.56.153:9092", "10.0.56.153:9093", "10.0.56.153:9094"},"testgroup", consumerConfig)

    if err != nil {

       panic(err)

    }

   for {

      err := cGroup.Consume(context.Background(), []string{"testAutoSyncOffset"}, consumerGroup)

      if err != nil {

         fmt.Println(err.Error())

         break

      }

   }

   _ = cGroup.Close()

}