func SpanFromMessage(ctx context.Context, tracer opentracing.Tracer, message *kafka.Message) (opentracing.Span, context.Context, error) {
	carrier := getCarrier(message)
	spanContext, err := tracer.Extract(opentracing.TextMap, carrier)
	if err != nil && err != opentracing.ErrSpanContextNotFound {
		return nil, nil, err
	}
	span := tracer.StartSpan("kafka reader", ext.RPCServerOption(spanContext))
	if err != nil && err != opentracing.ErrSpanContextNotFound {
		return nil, nil, err
	}
	ext.SpanKind.Set(span, ext.SpanKindConsumerEnum)
	ext.PeerService.Set(span, "kafka")
	span.SetTag("topic", message.Topic)
	span.SetTag("partition", message.Partition)
	span.SetTag("offset", message.Offset)
	return span, opentracing.ContextWithSpan(ctx, span), nil
}
SpanFromMessage reads the message
package otkafka
import (
	"context"
	"github.com/opentracing/opentracing-go"
	"github.com/opentracing/opentracing-go/ext"
	"github.com/segmentio/kafka-go"
)