Injecting Baggage into Kafka Message Headers

Build a carrier with the OpenTelemetry propagator, encode each entry to UTF-8 bytes, and append it to the record’s headers list at produce time — the SDK writes both traceparent and baggage into that carrier only if W3CBaggagePropagator is in your global propagator chain.

Context & When It Matters

Kafka records carry an optional list of headers — arbitrary (key, value) pairs stored beside the key and payload. This is the natural home for propagated context on an async transport, because unlike HTTP there is no ambient request object to hang headers off. When you want a routing signal like tenant or canary to survive from a producer through a topic to a downstream consumer, it has to be serialized into those record headers explicitly. Getting the trace traceparent there is well-trodden; getting the baggage header there is where teams slip, because a propagator chain missing the baggage propagator produces no error — the header is simply never written. This page covers only the write side and its immediate consumer read; the full Kafka and gRPC guide covers the broader picture including gRPC.

The reason baggage is worth the trouble on Kafka specifically is that a topic is a decoupling point. The producer and consumer may be owned by different teams, deployed on different schedules, and separated by minutes or hours of broker retention. A consumer that needs to know which tenant a message belongs to has three options: re-derive it from the payload (which couples the consumer to the message schema), re-query a database (which reintroduces the latency and coupling baggage exists to remove), or read it from a header the producer already knew. The header is almost always the cheapest and most honest choice, and it is the only one that keeps the routing signal aligned with the trace that carries it. That alignment is what lets a downstream router send canary traffic to a canary pool without re-deriving cohort membership from scratch.

Minimal Working Example First

Here is the smallest end-to-end that injects baggage on produce and reads it back on consume, using aiokafka and asyncio.

import asyncio
from aiokafka import AIOKafkaProducer, AIOKafkaConsumer
from opentelemetry import baggage
from opentelemetry.propagate import inject, extract, set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator

# Baggage moves ONLY because W3CBaggagePropagator is in this chain.
set_global_textmap(CompositePropagator([
    TraceContextTextMapPropagator(),
    W3CBaggagePropagator(),
]))

async def produce():
    producer = AIOKafkaProducer(bootstrap_servers="localhost:9092")
    await producer.start()
    ctx = baggage.set_baggage("tenant", "acme")
    ctx = baggage.set_baggage("canary", "true", context=ctx)

    carrier: dict[str, str] = {}
    inject(carrier, context=ctx)                      # writes traceparent + baggage
    headers = [(k, v.encode("utf-8")) for k, v in carrier.items()]  # str -> bytes
    await producer.send_and_wait("orders", b"payload", headers=headers)
    await producer.stop()

async def consume():
    consumer = AIOKafkaConsumer("orders", bootstrap_servers="localhost:9092",
                                group_id="settlement", auto_offset_reset="earliest")
    await consumer.start()
    async for msg in consumer:
        carrier = {k: v.decode("utf-8") for k, v in (msg.headers or [])}  # bytes -> str
        ctx = extract(carrier)
        print(baggage.get_baggage("tenant", context=ctx))   # -> "acme"
        break
    await consumer.stop()

The load-bearing lines are the two comprehensions: v.encode("utf-8") on the way in and v.decode("utf-8") on the way out. Everything else is standard client setup.

Implementation

A production producer wraps injection in a span so the send itself is traced, sets only the keys it owns, and keeps the header-building logic in one place. The annotated version below is what you would actually ship.

from aiokafka import AIOKafkaProducer
from opentelemetry import baggage, trace
from opentelemetry.propagate import inject

tracer = trace.get_tracer("orders.producer")

async def publish_order(producer: AIOKafkaProducer, order_id: str, payload: bytes):
    # Branch a fresh context for THIS message. set_baggage returns a new Context;
    # it never mutates the parent, so concurrent sends don't clobber each other.
    ctx = baggage.set_baggage("tenant", current_tenant())
    ctx = baggage.set_baggage("canary", "true" if in_canary() else "false", context=ctx)

    with tracer.start_as_current_span("order.publish", context=ctx) as span:
        carrier: dict[str, str] = {}
        # inject() reads the context you pass; giving it ctx (not the active one)
        # guarantees the baggage set above is what lands in the carrier.
        inject(carrier, context=ctx)                 # -> {"traceparent": ..., "baggage": ...}

        # Kafka headers MUST be (str, bytes). One un-encoded value raises TypeError.
        headers = [(key, value.encode("utf-8")) for key, value in carrier.items()]

        span.set_attribute("messaging.system", "kafka")
        span.set_attribute("messaging.destination.name", "orders")
        # Optionally copy the routing key into a span attribute so it's queryable
        # in the backend as well as propagated on the wire.
        span.set_attribute("tenant.id", baggage.get_baggage("tenant", context=ctx))

        await producer.send_and_wait("orders", key=order_id.encode(), value=payload,
                                     headers=headers)

Two details make this correct rather than merely working. First, inject(carrier, context=ctx) is passed the explicit context — under concurrent asyncio sends the active context may belong to a different in-flight request, and passing ctx removes that ambiguity. Under a single event loop, dozens of publish_order coroutines can be in flight at once; each one built its own ctx from set_baggage, and because set_baggage returns a fresh immutable Context rather than mutating a shared one, those branches never interfere. Passing the branch you built directly to inject is what preserves that isolation all the way to the wire. Second, the routing key is copied into a span attribute; baggage travels on the wire for routing, while the attribute makes the same value queryable in the tracing backend, which is the standard division between baggage and span attributes.

If you produce from a synchronous confluent-kafka client instead, the injection logic is identical — the only differences are that producer.produce(...) is fire-and-forget and you call producer.poll(0) or flush() to drive delivery callbacks. The carrier-to-bytes encoding does not change, because both clients model headers the same way: a list of (str, bytes) tuples. Keep the encoding in a single helper shared across your producers so a future client swap cannot silently reintroduce the string-versus-bytes bug.

Verification & Decision Checklist

Confirm the header actually landed before trusting downstream routing:

Use baggage-in-headers when a downstream consumer needs a routing or tenant signal that is not in the payload and you do not want it to re-query a database to recover it. If the value is only needed for post-hoc analysis, skip baggage and write a span attribute at the producer instead — it is cheaper because it is not paid per message on the wire.

One more verification worth automating: assert that the only baggage keys on the wire are the ones you intended. Baggage accumulates — a key set three services upstream rides along by default — so a producer can unintentionally forward internal keys it never meant to publish onto a shared topic. A short test that reads a produced record’s baggage header and checks its key set against an allowlist catches this before an unexpected key leaks to a consumer owned by another team.

Common Pitfalls

  • Headers as bytes, not strings. aiokafka and confluent-kafka both require bytes header values. Forgetting .encode() on inject or .decode() on extract is the most common failure; the encode side raises immediately, but a missed decode silently yields a carrier of bytes values that the propagator cannot parse, so extract() returns empty baggage.
  • Re-consumption creating duplicate root spans. A replay or offset reset re-reads records whose traceparent points at a long-finished trace. Extracting it as parent produces an orphaned span linked to a possibly-dropped parent. For replays, start a new trace and add a span link to the original rather than adopting its traceparent.
  • Baggage size added to every message. Unlike an HTTP request where headers are paid once per call, a baggage payload is stored, replicated, and re-read for every record in the topic. A careless 500-byte payload across a high-throughput topic is gigabytes of header storage per day. Budget keys deliberately; the size limits and header constraints guide quantifies the ceilings.

FAQ

Do I need to encode baggage values when writing Kafka headers?

Yes. Kafka record headers are (str, bytes) pairs, so every value the OpenTelemetry carrier produces must be UTF-8 encoded before it goes into the header list, and decoded back to str on the consumer side. Passing a raw string raises a type error in aiokafka and confluent-kafka alike.

Why does re-consuming a Kafka topic create duplicate root spans?

A replay tool or a consumer that resets its offset re-reads records whose baggage and traceparent headers still point at the original, long-finished trace. Extracting that stale traceparent makes the new consumer span a child of a span the backend may have already dropped, so it renders as an orphan root. Start a fresh trace for replays, or link to the original instead of extracting it as parent.

↑ Back to Propagating Baggage Across Kafka and gRPC