RecordDecoder

Turns a committable Kafka message into a DecodedRecord, inside a CONSUMER span.

Total by construction. Every failure path — an unrecognised content mode, malformed JSON, a missing required attribute, an envelope with no time and therefore no partition to land in, and even an unexpected runtime exception — becomes a Left(DeadLetter). Nothing here throws, because the only thing a throw could achieve is to kill a stream whose restart would meet the same record again.

Trace continuation happens here (ADR §7.2). KafkaTrace.withConsumerSpan extracts the W3C traceparent the producer injected and opens a CONSUMER span parented to it, so a trace that starts at wolfram's HTTP ingress continues into this consumer rather than starting a second, unrelated root trace. The span is made current for the duration of the decode, which is what puts trace_id/span_id into the MDC of anything logged from it.

Known limitation, stated rather than hidden: the span covers the decode, not the batched insert. The insert is a batch operation spanning many traces at once, so it cannot be a child of any one of them; it gets its own span in BatchProcessor. Holding every record's span open across the batch boundary was the alternative and it makes the consumer hold one live span per in-flight record — up to batchSize of them — with no way to end them if the stream is torn down mid-batch.