Propagating Baggage Across Kafka and gRPC

A payments team turned on canary routing keyed on a tenant and canary baggage entry and it worked perfectly over HTTP. Then the request crossed an async boundary: an order service published to a Kafka topic, a settlement worker consumed it, and that worker called a ledger service over gRPC. On the far side, the canary flag was gone. The traceparent still linked the spans into one trace, so the trace looked healthy — but the routing metadata that rides beside it had silently evaporated, and every settlement landed on the stable pool regardless of cohort.

This is the defining trap of non-HTTP propagation: trace context and baggage are two different headers carried by two different propagators, and it is entirely possible to configure one without the other. Auto-instrumentation for Kafka and gRPC frequently wires up traceparent and stops there. This guide fixes that end to end — injecting and extracting the W3C baggage header across Kafka message headers and gRPC metadata, keeping it alive through the consumer poll loop, and verifying it lands where routing decisions are made.

Prerequisites

  • OpenTelemetry SDK 1.20+ for Python (opentelemetry-sdk, opentelemetry-api) with the baggage propagator available under opentelemetry.baggage.propagation.
  • A Kafka client — confluent-kafka 2.x or aiokafka 0.10+ — plus grpcio 1.60+ for the gRPC examples. Go examples target go.opentelemetry.io/otel 1.24+.
  • A working OpenTelemetry SDK setup with an exporter already sending spans to a backend, so you can confirm baggage-derived attributes downstream.
  • Familiarity with the inject/extract lifecycle described in the baggage and metadata routing overview, and with the difference between baggage and span attributes.
  • Kafka broker access to inspect message headers (for example kafka-console-consumer --property print.headers=true).

Why Baggage Needs Its Own Propagator on Non-HTTP Transports

The W3C specification defines two independent concerns. traceparent/tracestate carry causal identity — the trace ID and parent span that stitch spans together. The baggage header carries application key-value pairs that ride alongside for routing and enrichment. In OpenTelemetry these are handled by two distinct propagators: TraceContextTextMapPropagator and W3CBaggagePropagator. A CompositePropagator chains them so a single inject() writes every header and a single extract() restores every piece of context.

Over HTTP this composition is usually configured once and forgotten. On Kafka and gRPC the danger is that instrumentation authors — including some auto-instrumentation packages — register only the trace-context propagator, because that is all a trace needs to render. Baggage is invisible in the trace waterfall, so its absence produces no broken-trace symptom. The routing logic simply fails quietly.

Two properties of the CompositePropagator are worth internalising before you wire anything up. First, order is not significant for correctness: each child propagator reads and writes its own header keys, so TraceContext never contends with W3CBaggage over the same bytes. You can list them in either order. Second, the composition is symmetric across the fleet — the propagator set on the producer determines which headers are written, and the set on the consumer determines which are read, and they must agree. A producer that injects baggage into a topic whose consumers lack the baggage propagator will emit perfectly valid headers that the downstream simply ignores. In a polyglot estate mixing Python, Go, and Java services, the single most reliable habit is to build the same composite in a shared bootstrap module per language and import it before the first produce, consume, or RPC call fires.

The concept diagram below shows the shape of the problem and the fix: on both transports the same two logical headers must travel together, mapped onto that transport’s native carrier — Kafka record headers or gRPC metadata.

Baggage and traceparent across Kafka headers and gRPC metadata Top row: a producer injects traceparent and baggage into Kafka record headers, the broker stores them, and the consumer extracts both. Bottom row: a gRPC client injects traceparent and baggage into request metadata and the server extracts both. Both rows emphasise that baggage travels as a second header beside traceparent. Kafka transport Producer inject() → carrier carrier → headers traceparent baggage Kafka Topic record headers stored traceparent baggage Consumer headers → carrier extract() + attach gRPC transport gRPC Client interceptor injects carrier → metadata metadata: traceparent metadata: baggage gRPC Server interceptor extracts attach → handler Both transports: baggage is a SECOND header beside traceparent — it only travels if W3CBaggagePropagator is in the propagator chain.

The amber labels mark the two headers that must move as a pair. If you take one architectural idea from this guide, it is that baggage is never implied by trace context — it is a separate payload that each transport must be told to carry.

Step-by-Step Implementation

Step 1: Register a CompositePropagator that includes baggage

Do this once, globally, in every service — producer, consumer, gRPC client, and gRPC server alike. The chain must contain W3CBaggagePropagator, or nothing that follows will move baggage.

# propagators.py — import this module before any produce/consume/RPC call
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import W3CBaggagePropagator

# TraceContext handles traceparent/tracestate; W3CBaggagePropagator handles baggage.
# Both must be present, or inject()/extract() will silently skip the baggage header.
set_global_textmap(CompositePropagator([
    TraceContextTextMapPropagator(),
    W3CBaggagePropagator(),
]))

With the global propagator set, opentelemetry.propagate.inject() writes both headers into any carrier and extract() restores both. Every step below relies on this being in place.

Step 2: Inject baggage at the Kafka producer

Kafka message headers are a list of (key, bytes) tuples. The SDK writes strings into a dict carrier; you then copy the carrier into the record’s header list as UTF-8 bytes.

from opentelemetry import baggage, context, trace
from opentelemetry.propagate import inject
from confluent_kafka import Producer

tracer = trace.get_tracer(__name__)
producer = Producer({"bootstrap.servers": "localhost:9092"})

def publish_order(order_id: str, payload: bytes):
    # Set routing baggage on a fresh context branch for this message.
    ctx = baggage.set_baggage("tenant", "acme")
    ctx = baggage.set_baggage("canary", "true", context=ctx)

    with tracer.start_as_current_span("order.publish", context=ctx):
        carrier: dict[str, str] = {}
        # inject() reads the ACTIVE context; pass ctx explicitly to be safe.
        inject(carrier, context=ctx)          # writes traceparent AND baggage
        # Kafka headers are (str, bytes) — encode every carrier value.
        headers = [(k, v.encode("utf-8")) for k, v in carrier.items()]
        producer.produce("orders", key=order_id, value=payload, headers=headers)
    producer.flush()

The record now leaves the producer with a baggage header reading tenant=acme,canary=true beside the traceparent.

Step 3: Extract and attach baggage per record at the consumer

Extraction alone does not change the active context. You must attach the extracted Context for each record and detach it afterward, otherwise the baggage stays inert. Doing this per record — never per batch — prevents one message’s baggage from bleeding into the next.

from opentelemetry import baggage, context as otel_context, trace
from opentelemetry.propagate import extract
from confluent_kafka import Consumer

tracer = trace.get_tracer(__name__)
consumer = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "settlement",
    "auto.offset.reset": "earliest",
})
consumer.subscribe(["orders"])

while True:
    record = consumer.poll(1.0)
    if record is None or record.error():
        continue

    # Decode Kafka's (bytes, bytes) headers into a string carrier.
    carrier = {k: v.decode("utf-8") for k, v in (record.headers() or [])}
    ctx = extract(carrier)                     # rebuilds traceparent AND baggage

    token = otel_context.attach(ctx)           # baggage now live on this thread
    try:
        with tracer.start_as_current_span("order.settle"):
            tenant = baggage.get_baggage("tenant")   # "acme"
            canary = baggage.get_baggage("canary")   # "true"
            settle(record.value(), route_canary=(canary == "true"))
    finally:
        otel_context.detach(token)             # prevent leakage into next record

Step 4: Inject baggage in a gRPC client interceptor

For gRPC the carrier is call metadata. A UnaryUnaryClientInterceptor injects the active context into a metadata list before the call goes out. This mirrors service mesh propagation, except the interceptor lives inside the application rather than in a sidecar.

import grpc
from opentelemetry.propagate import inject

class BaggageClientInterceptor(grpc.UnaryUnaryClientInterceptor):
    def intercept_unary_unary(self, continuation, client_call_details, request):
        carrier: dict[str, str] = {}
        inject(carrier)                        # active context → traceparent + baggage
        metadata = list(client_call_details.metadata or [])
        # gRPC metadata keys are lowercase ASCII; baggage value is printable ASCII.
        metadata.extend((k.lower(), v) for k, v in carrier.items())
        new_details = client_call_details._replace(metadata=metadata)
        return continuation(new_details, request)

channel = grpc.intercept_channel(
    grpc.insecure_channel("ledger:50051"),
    BaggageClientInterceptor(),
)

Step 5: Extract baggage in a gRPC server interceptor

On the server side, a ServerInterceptor reads the invocation metadata into a carrier, extracts the Context, and attaches it so the handler and every downstream call observe the baggage.

import grpc
from opentelemetry import context as otel_context
from opentelemetry.propagate import extract

class BaggageServerInterceptor(grpc.ServerInterceptor):
    def intercept_service(self, continuation, handler_call_details):
        # invocation_metadata is a tuple of (key, value) string pairs.
        carrier = dict(handler_call_details.invocation_metadata or ())
        ctx = extract(carrier)                 # rebuild traceparent + baggage

        handler = continuation(handler_call_details)

        def wrap_unary(request, servicer_context):
            token = otel_context.attach(ctx)   # baggage live for the handler
            try:
                return handler.unary_unary(request, servicer_context)
            finally:
                otel_context.detach(token)

        return grpc.unary_unary_rpc_method_handler(
            wrap_unary,
            request_deserializer=handler.request_deserializer,
            response_serializer=handler.response_serializer,
        )

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=8),
    interceptors=[BaggageServerInterceptor()],
)

For Go, the equivalent uses otelgrpc stats handlers with a global propagator built from propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}); the Baggage{} propagator is the Go analogue of W3CBaggagePropagator and must be included explicitly for the same reason.

Verification

Confirm baggage survives each hop rather than trusting that the trace rendered.

First, inspect the raw Kafka headers to prove the producer wrote a baggage header:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic orders --property print.headers=true --max-messages 1
# Expect a header line containing: baggage:tenant=acme,canary=true

Second, assert inside the consumer and the gRPC handler that the keys are readable:

from opentelemetry import baggage

# Drop this into the consumer span or gRPC handler during a test run.
assert baggage.get_baggage("tenant") == "acme"
assert dict(baggage.get_all()).get("canary") == "true"

Third, confirm the routing signal reaches the terminal service. If you copy tenant into a span attribute at each hop, a backend query proves the value crossed both transports — search for the trace in Jaeger or Tempo and check that the settlement and ledger spans both carry tenant=acme. A missing attribute on the ledger span pinpoints exactly which transport dropped the baggage.

This copy-to-attribute step is also the cheapest permanent guardrail. Baggage itself never lands in the backend — it lives only on the wire — so the only durable evidence that propagation worked is an attribute you deliberately materialise. Add it at the ingress of each service (the Kafka consumer span and the gRPC handler span) rather than deep in business logic, and you get a per-hop breadcrumb trail: the first hop where the attribute goes missing is the hop whose propagator chain is misconfigured. Keep the attribute low-cardinality — a tenant slug or a boolean canary flag, never a raw user identifier — so you do not trade a propagation bug for a cardinality explosion in the metrics derived from spans.

Edge Cases & Gotchas

  1. Broker header byte accounting. Every header is stored and replicated per message. A 400-byte baggage payload across ten million daily messages is 4 GB of header data per day before replication. Baggage almost never trips the per-record message.max.bytes ceiling, but it is not free — treat header bytes as a recurring cost, not a one-time header.
  2. Binary vs string gRPC metadata keys. gRPC treats any key ending in -bin as binary and base64-encodes its value. The baggage value is printable ASCII and must use a plain string key. Naming it baggage-bin breaks interop with standard OpenTelemetry gRPC instrumentation, which looks for the string key baggage.
  3. Consumer poll-loop context loss. The poll loop thread carries whatever context was active before poll() returned — typically empty. Extraction does not fix this; you must attach() per record. This is the single most common reason baggage “disappears” on the consumer side even though the header is present on the wire.
  4. Reserved gRPC metadata keys. Keys beginning with grpc- are reserved by the framework. Never rename baggage to a grpc--prefixed key; the runtime may strip or reject it.
  5. Batch consumption sharing one context. When poll() or a batch handler returns many records, attaching the first record’s context for the whole batch tags every sibling message with the first message’s baggage. Extract and attach inside the per-record loop body.
  6. Streaming RPC context reuse. For a long-lived gRPC stream, metadata is sent once at stream open. Baggage captured at that moment is frozen for the stream’s lifetime; per-message baggage changes will not propagate. Use unary calls when baggage must vary per message.

Performance & Scale Notes

Injecting and extracting the composite propagator adds roughly 2–5 µs per Kafka message and per gRPC call — dominated by the string encoding of the baggage header, not by trace context. At 100k messages/second this is well under 0.5% CPU, comfortably below serialization and business-logic cost. The scale concern is not CPU but bytes: baggage is paid on every message, so a disciplined key budget (a handful of short keys, values under ~50 bytes) matters far more here than on a low-volume HTTP path. For guidance on those budgets and the transport ceilings they run into, see baggage size limits and header constraints. Async boundaries inside consumers — thread pools, asyncio tasks — need the same async context handling as any other propagation path; the attach/detach discipline above is what keeps baggage from leaking across pooled workers.

Troubleshooting FAQ

Why does traceparent propagate across Kafka but baggage does not?

Because a bare TraceContextTextMapPropagator only handles the traceparent and tracestate headers. Baggage is written and read by a separate W3CBaggagePropagator. If your propagator chain omits it, inject() never emits the baggage header and extract() never restores baggage entries. Register a CompositePropagator that includes W3CBaggagePropagator on both the producer and the consumer.

Do Kafka message headers have a size limit that baggage can exceed?

Yes. Kafka bounds each record against message.max.bytes (default about 1 MB), and headers count toward that total. Baggage rarely hits the record ceiling, but every header byte is stored per message and replicated, so a 400-byte baggage payload multiplied by millions of messages adds real storage and network cost. Keep baggage to a few short keys and monitor total header size.

Should gRPC baggage metadata keys use the -bin suffix?

No. The W3C baggage header value is printable ASCII, so it maps cleanly to a normal string metadata key. Reserve the -bin suffix for binary values that gRPC base64-encodes on the wire. Using -bin for baggage forces needless encoding and breaks the standard OpenTelemetry gRPC instrumentation, which expects string keys named traceparent, tracestate, and baggage.

Why does baggage disappear inside my Kafka consumer poll loop?

The poll loop runs on a thread whose active context is whatever was set before poll() returned — usually empty. Extraction alone does not change the active context; you must attach the extracted Context per record and detach it in a finally block. Sharing one attached context across a whole batch bleeds one record’s baggage into every sibling message.

↑ Back to Baggage & Metadata Routing Workflows