如何使用Jaeger追踪Kafka生产者和消费者

Question

I want to tracing kafka producer and consumer by using jaeger and Go, but I can not get the chain of the span for the producer's tracer and consumer's tracer. Here is the producer's code:

func (p *Producer) WriteMessages(ctx context.Context, messages ...kafka.Message) error {
    topic := p.Stats().Topic
    for _, message := range messages {
        // span := opentracing.GlobalTracer().StartSpan("TO_"+topic, ext.SpanKindProducer)
        span, _ := opentracing.StartSpanFromContext(ctx, "TO_"+topic, ext.SpanKindProducer)
        ext.Component.Set(span, "golang-kafka")
        ext.PeerService.Set(span, "kafka")
        ext.MessageBusDestination.Set(span, topic)
        headers := make(map[string]string)
        opentracing.GlobalTracer().Inject(
            span.Context(),
            opentracing.TextMap,
            opentracing.TextMapCarrier(headers),
        )
        for key, value := range headers {
            header := kafka.Header{
                Key:   key,
                Value: []byte(value),
            }
            message.Headers = append(message.Headers, header)
        }
        span.Finish()
    }

    return p.Writer.WriteMessages(ctx, messages...)
}

and the consumer's code:

func (c *Consumer) ReadMessage(ctx context.Context) (kafka.Message, error) {
    message, err := c.Reader.ReadMessage(ctx)
    if err != nil {
        return kafka.Message{}, err
    }
    topic, partition, offset := message.Topic, message.Partition, message.Offset
    headers := make(map[string]string)
    for _, header := range message.Headers {
        headers[header.Key] = string(header.Value)
    }
    spanContext, _ := opentracing.GlobalTracer().Extract(
        opentracing.TextMap,
        opentracing.TextMapCarrier(headers),
    )
    span, _ := opentracing.StartSpanFromContext(ctx, "FROM_"+topic, opentracing.FollowsFrom(spanContext), ext.SpanKindConsumer)
    // span := opentracing.StartSpan("FROM_"+topic, opentracing.FollowsFrom(spanContext), ext.SpanKindConsumer)
    ext.Component.Set(span, "golang-kafka")
    ext.PeerService.Set(span, "kafka")
    span.SetTag("topic", topic)
    span.SetTag("partition", partition)
    span.SetTag("offset", offset)
    span.Finish()

    return message, err
}

I can get the producer chain and the consumer chain, but I can not chain them together.