Python Observability

Tracing Kafka Producers and Consumers in Python

Kafka decouples producers from consumers in time, which is its purpose and the reason tracing across it needs care. Trace context has to travel inside each message, in its headers; the producer's span measures handing the message to a buffer rather than the broker accepting it; and a consumer may process the message seconds or hours later, possibly in a batch alongside messages from many different requests. This page covers each of those. It is a task article under tracing databases and message queues, part of the distributed tracing and OpenTelemetry in Python section.

Spans on both sides, context in between A request in an orders service publishes an order event. The producer span starts when send is called, injects the trace context into the message headers, and ends as soon as the message is placed in the client's local buffer. Later, the client's background thread delivers the batch to the broker, and a delivery callback reports the partition and offset, or an error. The message sits in the topic for forty seconds. A consumer in a fulfilment service polls it, extracts the context from the headers, and starts a consumer span related to the producer span, recording the queue wait of forty seconds as an attribute. The consumer's own database spans are children of the consumer span. The note records that the producer span's short duration and the consumer's queue wait together describe the message's journey, and that neither alone does. one message, two services, forty seconds apart orders (producer) POST /orders send — 0.2 ms delivery callback later: partition 3, offset 88120 headers order-events waits 40 s fulfilment (consumer) process order-events queue_wait_ms = 40112 UPDATE … the producer span measures the enqueue; the callback reports delivery the consumer span measures processing; the queue wait attribute measures the gap together they describe the message's journey — neither does alone
Each side records its own part of the journey, and the headers carry the context between them. The queue wait is usually the largest number and the one most often unrecorded.

Prerequisites

pip install "confluent-kafka>=2.3.0,<3.0.0" \
            "opentelemetry-instrumentation-confluent-kafka>=0.48b0,<1.0.0" \
            "opentelemetry-sdk>=1.27.0,<2.0.0"

Implementation

Step 1 — Instrument both clients. The instrumentation wraps producer and consumer instances. The producer wrapper creates a producer span for each send and injects the current context into the message headers; the consumer wrapper extracts context from each polled message and creates a consumer span. Both sides must be instrumented, and nothing between them may strip headers.

from confluent_kafka import Producer, Consumer
from opentelemetry.instrumentation.confluent_kafka import ConfluentKafkaInstrumentor

inst = ConfluentKafkaInstrumentor()
producer = inst.instrument_producer(Producer({"bootstrap.servers": BROKERS}))
consumer = inst.instrument_consumer(Consumer({
    "bootstrap.servers": BROKERS,
    "group.id": "fulfilment",
    "enable.auto.commit": False,
}))

Step 2 — Record the delivery outcome from the callback. Producing is asynchronous: produce places the message in a local buffer and returns immediately, and the broker's acknowledgement arrives later through a delivery callback. The producer span therefore measures the enqueue — usually a fraction of a millisecond — and says nothing about whether the message was delivered. Recording the partition, offset or error from the callback, as an event or a log record carrying the same trace identifier, completes the picture.

from opentelemetry import trace
import logging

log = logging.getLogger("orders.events")

def publish_order_event(order):
    span_ctx = trace.get_current_span().get_span_context()

    def on_delivery(err, msg):
        if err is not None:
            log.error("event delivery failed", extra={
                "trace_id": f"{span_ctx.trace_id:032x}", "kafka_error": str(err)})
        else:
            log.debug("event delivered", extra={
                "trace_id": f"{span_ctx.trace_id:032x}",
                "partition": msg.partition(), "offset": msg.offset()})

    producer.produce("order-events", key=order.id, value=serialise(order),
                     on_delivery=on_delivery)
    producer.poll(0)                               # serve delivery callbacks

Step 3 — Record the queue wait on the consumer span. The difference between the message's timestamp and the moment processing starts is how long the message waited in the topic. It is frequently the largest component of end-to-end latency and it is invisible in the consumer span's own duration. Adding it as an attribute makes it queryable, and it is the per-message counterpart of consumer group lag.

import time

def handle(msg):
    span = trace.get_current_span()                # the consumer span from the wrapper
    ts_type, ts_ms = msg.timestamp()
    if span.is_recording() and ts_ms > 0:
        span.set_attribute("app.queue_wait_ms", int(time.time() * 1000) - ts_ms)
        span.set_attribute("messaging.kafka.message.offset", msg.offset())
        span.set_attribute("messaging.kafka.destination.partition", msg.partition())
    fulfil(msg.value())

Step 4 — Choose how consumer spans relate to producers. With the default arrangement, the consumer span is a child of the producer span, so one trace covers request, publish and processing — and that trace's duration includes the queue wait. The alternative is a new root for the consumer with a link to the producer, which keeps each trace's duration meaningful. Either is valid; mixing them across a fleet is what causes confusion. Tracing databases and message queues discusses the trade.

Step 5 — Link batches to every producer. Consumers that poll and process many messages together cannot sensibly be the child of one producer. A span for the batch, with a link to each message's extracted context, relates the batch to every request that contributed to it, as described in adding span links for batch work.

from opentelemetry.propagate import extract
from opentelemetry.trace import Link, SpanKind

tracer = trace.get_tracer("fulfilment")

def process_batch(messages):
    links = []
    for m in messages[:64]:
        headers = {k: v.decode() for k, v in (m.headers() or [])}
        ctx = trace.get_current_span(extract(headers)).get_span_context()
        if ctx.is_valid:
            links.append(Link(ctx, {"messaging.kafka.message.offset": m.offset()}))
    with tracer.start_as_current_span("process order-events batch",
                                      kind=SpanKind.CONSUMER, links=links) as span:
        span.set_attribute("messaging.batch.message_count", len(messages))
        for m in messages:
            fulfil(m.value())

Expected Output: a consumer span with its queue wait, related to the producer.

process order-events  CONSUMER  204 ms  app.queue_wait_ms=40112  partition=3 offset=88120
  UPDATE shipments    CLIENT     18 ms
Lag and queue wait answer different questions Two measurements of the same consumer group are shown. Consumer lag, a metric, plots how many messages behind the latest offset the group is over an hour: flat near zero, then rising sharply during a traffic spike and draining over twenty minutes. Queue wait, an attribute on each consumer span, gives the per-message delay: messages consumed during the drain carry queue waits of up to eighteen minutes. The lag metric says the group fell behind and when; the queue wait on a specific trace says how late a specific order was processed, which is what a support ticket about that order needs. The note records that lag belongs in metrics for alerting, and queue wait belongs on spans for explaining individual cases. one consumer group, two views of the same backlog lag (metric) fell behind, drained over 20 min queue wait (span) order ord_7: processed 18 min late lag in metrics for alerting · queue wait on spans for explaining one order a support ticket about one late order needs the second; an on-call page needs the first
Lag says the group fell behind. Queue wait says how late one particular message was handled, which is the question an individual complaint actually asks.

Headers, bridges and the places context gets lost

Kafka stores headers with every message, so context survives however long a message waits. It does not survive everything between producer and consumer, and the common failure points are worth checking when consumer traces stop relating to producers.

Stream processors and bridges. A component that consumes from one topic and produces to another — a stream processing job, a connector, a mirroring tool — must copy headers from input to output. Many do not by default. The downstream consumer then sees messages with no trace context and starts fresh traces. Configuring the component to preserve headers, or injecting new context that links to the extracted one, restores the chain.

Serialisation frameworks that rebuild messages. Some producer wrappers construct a new message object from a schema and do not carry arbitrary headers through. Checking a message on the topic with a console consumer that prints headers confirms whether the traceparent is present.

Consumers that read headers wrongly. Header values are bytes, and a consumer that extracts from them without decoding — or a hand-written extraction that expects a different header name — silently fails to find context. The instrumentation handles this correctly; custom consumer loops are where it breaks.

Retries through dead-letter topics. A message moved to a retry or dead-letter topic by an error handler should keep its original headers, so the eventual reprocessing relates to the original request. Handlers that republish only the value lose it.

For each, the check is the same: read a message from the topic in question and look at its headers. A traceparent present on the input topic and absent on the output names the component responsible.

Commit, retry and the consumer's own failures

The consumer span describes processing one message. Two surrounding behaviours determine whether that span tells the whole story.

Offset commits. With auto-commit disabled, as above, the consumer commits offsets after processing. A crash between processing and commit means the message is processed again after restart, producing a second consumer span for the same message — possibly in a different trace if the consumer is a new root per message. Recording the offset as an attribute, as step 3 does, lets duplicates be recognised: two consumer spans with the same partition and offset are the same message processed twice. That is the evidence needed when a downstream effect happens twice and someone asks why.

Retries within the consumer. A consumer that retries a failing message in place produces either one long span covering all attempts or several spans, depending on where the span begins. Starting the span per attempt, with an attempt number attribute, makes a flapping dependency visible; starting it once around the retry loop makes the total delay visible. Either works if applied consistently, and the attempt count belongs on the span in both cases.

Poison messages. A message that fails every time blocks a partition if the consumer keeps retrying it, and the symptom is lag rising on one partition while others are healthy. The failing consumer spans carry the offset and the exception; alerting on repeated failure spans for the same offset catches a poison message far sooner than lag alone, which rises slowly.

Which clock measures what A table of four time measurements around one Kafka message and what each answers. The producer span duration measures how long the send took to be acknowledged by the broker. The time between the producer span's end and the consumer span's start measures how long the message waited in the topic. Consumer lag in messages measures how far behind the group is, independent of any single message. The consumer span duration measures how long processing took. The note says end-to-end delay for users is the sum of the wait and the processing, and only tracing gives the per-message wait. measurement answers producer span duration how long until the broker acknowledged producer end → consumer start how long this message waited consumer lag (messages) how far behind the group is overall consumer span duration how long processing took delay users feel = time waiting + time processing only the trace gives the wait for an individual message
Lag counts messages; the trace times them. The per-message wait is visible only when context crosses the topic.

Configuration options

Concern Setting Why
Instrumentation both producer and consumer one-sided breaks the chain
Producer span measures the enqueue produce is asynchronous
Delivery outcome from the delivery callback the only confirmation
Queue wait attribute on the consumer span often the real latency
Consumer relationship child or linked root, fleet-wide predictable traces
Batches one span, a link per message many producers per batch
Headers preserved through every bridge context survives hops
Lag consumer group metric alerting on backlog

Verification

Produce and consume one message locally and confirm the consumer span relates to the producer.

spans = exporter.get_finished_spans()
prod = next(s for s in spans if s.kind == SpanKind.PRODUCER)
cons = next(s for s in spans if s.kind == SpanKind.CONSUMER)
related = (cons.parent and cons.parent.span_id == prod.context.span_id) or \
          any(l.context.span_id == prod.context.span_id for l in cons.links)
print("related:", related, "same trace:", cons.context.trace_id == prod.context.trace_id)

Expected Output: related, either as a child in the same trace or through a link.

related: True same trace: True

Common mistakes

Treating the producer span as delivery. Error signature: messages lost with no error on any span. Root cause: the send returns before the broker confirms. Remediation: record the delivery callback's outcome.

No queue wait attribute. Error signature: consumer traces that look fast for orders customers say were hours late. Root cause: waiting time is outside the consumer span. Remediation: record it from the message timestamp.

Headers dropped by a bridge. Error signature: consumers downstream of a stream processor never relate to producers. Root cause: the bridge republishes without headers. Remediation: preserve headers or inject linked context.

One producer as parent of a batch. Error signature: one request's trace stretched by an entire batch's processing. Root cause: parenting on the first message. Remediation: a batch span with links.

Never polling the producer. Error signature: delivery callbacks that never fire and a growing local buffer. Root cause: callbacks are served only by poll or flush. Remediation: poll regularly and flush on shutdown.

Frequently Asked Questions

How does trace context travel through Kafka?

In message headers. The producer instrumentation injects the traceparent and baggage into the headers of each message it sends; the consumer instrumentation extracts them. Kafka stores headers with the message, so context survives however long the message waits.

Why is the producer span so short?

Because producing is asynchronous. The send call places the message in the client's buffer and returns; the broker acknowledgement arrives later through a callback. The producer span measures the enqueue, and the delivery outcome must be recorded from the callback.

How do I see how long a message waited before being consumed?

Compare the message's timestamp with the time the consumer started processing it, and record the difference on the consumer span. That queue wait is often the latency a user actually experienced.

What about consumer lag?

Lag — how far behind the latest offset a consumer group is — is a property of the group over time and belongs in metrics. The queue wait on individual consumer spans is its per-message counterpart and explains specific slow traces.

Related Guides