func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		printMessage(message)
		session.MarkMessage(message, "")
	}
	return nil
}
ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
package main
import (
	"context"
	"flag"
	"log"
	"os"
	"strings"
	"time"
	"github.com/Shopify/sarama"
	"go.opentelemetry.io/otel"
	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
	"go.opentelemetry.io/otel/trace"
	"go.opentelemetry.io/contrib/instrumentation/github.com/Shopify/sarama/otelsarama"
	"go.opentelemetry.io/contrib/instrumentation/github.com/Shopify/sarama/otelsarama/example"
)
var (
	brokers = flag.String("brokers", os.Getenv("KAFKA_PEERS"), "The Kafka brokers to connect to, as a comma separated list")
)
func main() {
	tp := example.InitTracer()
	defer func() {
		if err := tp.Shutdown(context.Background()); err != nil {
			log.Printf("Error shutting down tracer provider: %v", err)
		}
	}()
	flag.Parse()
	if *brokers == "" {
		flag.PrintDefaults()
		os.Exit(1)
	}
	brokerList := strings.Split(*brokers, ",")
	log.Printf("Kafka brokers: %s", strings.Join(brokerList, ", "))
	startConsumerGroup(brokerList)
	select {}
}
func startConsumerGroup(brokerList []string) {
	consumerGroupHandler := Consumer{}
	handler := otelsarama.WrapConsumerGroupHandler(&consumerGroupHandler)
	config := sarama.NewConfig()
	config.Version = sarama.V2_5_0_0
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	consumerGroup, err := sarama.NewConsumerGroup(brokerList, "example", config)
	if err != nil {
		log.Fatalln("Failed to start sarama consumer group:", err)
	}
	err = consumerGroup.Consume(context.Background(), []string{example.KafkaTopic}, handler)
	if err != nil {
		log.Fatalln("Failed to consume via handler:", err)
	}
}
func printMessage(msg *sarama.ConsumerMessage) {
	ctx := otel.GetTextMapPropagator().Extract(context.Background(), otelsarama.NewConsumerMessageCarrier(msg))
	tr := otel.Tracer("consumer")
	_, span := tr.Start(ctx, "consume message", trace.WithAttributes(
		semconv.MessagingOperationProcess,
	))
	defer span.End()
	time.Sleep(1 * time.Second)
	log.Println("Successful to read message: ", string(msg.Value))
}
type Consumer struct {
}
func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error {
	return nil
}
func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error {
	return nil
}