package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/segmentio/kafka-go"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/trace"
  "go.opentelemetry.io/otel/attribute"
)

func main() {
	// 创建一个 OpenTelemetry 的 Tracer
	tracer := otel.Tracer("kafka-consumer")

	// 创建一个 Kafka 消费者
	consumer, err := kafka.NewConsumer(kafka.ConsumerConfig{
		Brokers: []string{"localhost:9092"},
		GroupID: "my-group",
		Topic:   "my-topic",
	})
	if err != nil {
		log.Fatal(err)
	}

	// 循环消费消息
	for {
		// 从 Kafka 中获取消息
		msg, err := consumer.ReadMessage(context.Background())
		if err != nil {
			log.Println("Error reading message:", err)
			continue
		}

		// 创建一个新的 Span
		ctx, span := tracer.Start(context.Background(), "process-message")
		defer span.End()

		// 在 Span 中添加 trace 信息
		span.SetAttributes(
			attribute.String("kafka.topic", msg.Topic),
			attribute.Int64("kafka.partition", int64(msg.Partition)),
			attribute.Int64("kafka.offset", int64(msg.Offset)),
		)

		// 处理消息
		fmt.Println("Received message:", string(msg.Value))

		// 结束 Span
		span.End()
	}

	// 关闭 Kafka 消费者
	consumer.Close()
}