W3C trace-context propagation across Kafka: inject on produce, extract on consume.
This is the whole cross-service tracing story (ADR §7.2). Without it a trace stops at wolfram's HTTP handler and restarts, unrelated, in cobalt — two disconnected traces where the interesting question ("why did this reading take 40 seconds to appear?") lives precisely in the gap between them. With it, one trace runs from HTTP ingestion, through the broker, into the consumer and its database write.
Why the propagator is a parameter with a W3C default rather than a hard-wired global.GlobalOpenTelemetry is exactly the sort of hidden singleton that makes a test's outcome depend on suite ordering. Services pass Telemetry.tracing.propagator; a test passes its own; the default keeps a service that has not wired telemetry yet from silently strippingtraceparent and breaking tracing for everyone downstream, which is what a no-op propagator would do.
Extraction always starts from Context.root(). The consumer's poll loop may still be carrying the previous record's context on that thread; inheriting it would chain every record in a batch into one ever-growing trace that says nothing about any single record.
-Werror trap (ADR §7.4). Kafka's Headers.add/remove and every OTel builder method return this, so writing them as statements fails under -Wnonunit-statement. They are chained or bound with val _ = below.
This module deliberately does not depend on com.worxbend.observability (ADR §3.3 lists eventing's dependencies as kernel, cloudevents, kafka-clients and opentelemetry-api). KafkaHeaderCarrier therefore exposes the three operations of that module's TextCarrier without naming it, so a module that sees both can bind given TextCarrier[Headers] in one delegating line and get Tracing.spanFrom over Kafka headers for free.
Reads trace context out of headers, starting from Context.root().
Reads trace context out of headers, starting from Context.root().
Returns root when there is no valid traceparent, which yields a fresh root trace rather than an error — the right behaviour for an event published by a producer that is not instrumented yet.
Writes the ambient trace context into headers, returning the same (mutable) headers.
Writes the ambient trace context into headers, returning the same (mutable) headers.
Ambient context is correct on wolfram's synchronous request path and wrong across every async boundary — a Future, a stream stage, a Vert.x event-loop hop. OTel's Context is ThreadLocal-backed and returns root rather than failing when it has been lost, so an orphaned span is silent. Use the explicit-context overload whenever a thread hop is involved.
Runs body inside a CONSUMER span that continues the trace the record carries.
Runs body inside a CONSUMER span that continues the trace the record carries.
The span is made current for the duration, so Span.current() — and therefore com.worxbend.observability.LogContext.withCurrentSpan in the consumer — sees it without the caller threading anything through. The MDC is deliberately not written here: the keys belong to modules/observability, and duplicating them in a module that cannot see them is how two spellings of trace_id end up in one log stream.
A thrown exception is recorded and the status set to ERROR before it is rethrown. Swallowing it would make instrumentation change program behaviour, which is the one thing instrumentation may never do; NonFatal only, because an OutOfMemoryError is not a span attribute.