com.worxbend.eventing

The one place where the domain com.worxbend.kernel.event.Envelope and the SDK's io.cloudevents.CloudEvent meet.

ADR §4 is emphatic that the SDK type is infrastructure, never a domain type: it is a Java interface with nullable getters, a throwing mutable builder and a byte-oriented data model. Keeping the conversion in a single object means wolfram (producer) and cobalt (consumer) cannot disagree about what a given envelope looks like on the wire — the whole reason modules/eventing exists at all.

Losslessness is stated as an equation, not asserted. The SDK cannot represent everything an Envelope can, so "lossless" is defined as: toEnvelope(toCloudEvent(e)) == Right(canonical(e)), with canonical an idempotent function that names every normalisation exactly once. Anything canonical does not change survives untouched; anything it does change was never representable, and the change is visible in one readable function rather than hidden across a codec. The property is tested for every generated envelope, and canonical is tested idempotent.

The three normalisations, and why each is forced:

• Reserved names are not extensions. id, time, data &c. are context attributes; an extension so named would collide with the attribute on the wire. They are dropped, matching Envelope.canonical.
• AttrValue.Other becomes Text. The SDK's extension model has exactly six types and a JSON array or object is not among them. Encoding the JSON text and decoding it back as a string keeps the data; guessing on the way back — "does this string parse as JSON?" — would corrupt every string extension that happens to look like JSON, which is the far more common case. The other five AttrValue shapes round-trip typed, which is the point of the ADT and is why this is not simply Map[String, String].
• The payload's shape follows the media type. CloudEventData is bytes; the data versus data_base64 distinction that kernel's JSON codec relies on does not exist here. So canonical resolves the shape by literally encoding and re-decoding the payload, which makes agreement between the two directions structural rather than a pair of functions someone has to keep in step.

Extension names cannot be normalised, so they are rejected. CloudEvents requires lowercase alphanumerics, and the SDK builder throws on anything else. Renaming would lose data and dropping would lose more, so toCloudEvent returns a Left naming the offending keys — which is exactly the ingest-time validation ADR §4.3 asks for ("Reject; never invent defaults") rather than a surprise RuntimeException inside a producer callback.

The Kafka header vocabulary of the CloudEvents protocol binding, and the three primitive operations over org.apache.kafka.common.header.Headers that everything else in this module is built from.

Why these names are re-declared here rather than imported from the SDK. io.cloudevents.kafka.impl.KafkaHeaders holds the same constants, but it lives in an impl package and CE_PREFIX is protected, so half of the vocabulary is unreachable from Scala anyway. Re-declaring them is only safe if the two definitions cannot drift apart silently — so CloudEventHeadersSuite asserts equality against the SDK's own public constants. That test is the contract; deleting it makes these strings a guess.

Values are UTF-8 text, not arbitrary bytes. The binding specifies every ce_* header and content-type as a string, so decoding a header value with UTF-8 is lossless for anything this system — or any conformant producer — emits. A third-party header carrying raw binary will still decode without throwing (UTF-8 substitutes replacement characters) and is recorded for diagnosis only; the record's value bytes, which is what a DLQ replay actually needs, are always kept verbatim.

The two ways the CloudEvents Kafka binding can lay an event onto a record, and the encode/decode pair for each.

Binary is the mode of events.cloudevents.v1; structured is the mode of the DLQ. ADR §4.3 picks both deliberately and for opposite reasons:

• Binary on the main topic puts the context attributes in headers and the payload in the value untouched. Brokers, single-message transforms and kcat route on ce_type / ce_source without deserializing anything; the payload is never re-encoded, so an event whose schema this build has never seen round-trips byte-identically; and binary data is not double-base64'd. The cost — extension attributes lose their type, because a header is bytes — is named and made explicit by CloudEventAdapter.binaryCanonical rather than discovered later.
• Structured on the DLQ puts the entire event, attributes and all, into the value as one self-contained application/cloudevents+json blob. A poison record is read by a human under time pressure with kcat, and reassembling an event from a dozen headers at that moment is exactly the wrong task. Self-containment also makes replay a copy rather than a reconstruction.

A single read handles both, because the consumer that drains the DLQ and the consumer that drains the main topic are the same code, and because a producer misconfigured into the other mode should be read rather than silently dead-lettered.

The wrapper that makes a poison pill diagnosable and replayable.

A record that cannot be decoded is the one case where the offset must be committed without the event having been processed (ADR §4.3). That is only defensible if nothing is lost, so this carries three things:

• why — DecodeFailure.reason, a bounded tag safe for consume.records.poison{reason}, next to the unbounded detail a human actually reads;
• where — the origin coordinates, which are the record's only identity when its id is the thing that failed to parse, and which double as the DLQ key so retries overwrite;
• what — the original value bytes verbatim, plus the original headers. Replay is republishing these two, so anything less makes the DLQ a graveyard rather than a queue.

Headers are recorded as text, the value as bytes. The binding defines every ce_* header as a UTF-8 string, so text is lossless for anything this system or a conformant producer writes, and it keeps the DLQ record readable in kcat — which is the entire reason the DLQ is in structured mode. The value is where arbitrary bytes actually live, and it is kept exactly, base64-encoded.

The dead letter is itself a CloudEvent. toEnvelope wraps it as a normal envelope with this build's own type, so the DLQ topic holds the same kind of thing as every other topic: it can be consumed by the same decoder, rendered by the same UI and stored by the same insert. A bespoke DLQ format would need a second reader that nobody exercises until the day it matters.

Why a Kafka record could not be turned into an com.worxbend.kernel.event.Envelope.

This is a value, never an exception. ADR §4.3 explains the failure it exists to prevent: a throwing deserializer throws inside KafkaConsumer.poll, before the stream connector ever sees the record. The stream dies, the offset is never committed, and the restart replays the same poison record forever — an outage that looks like a crash loop and whose cause is one malformed byte. Decoding therefore returns Either, the bad record becomes a DeadLetter, and the offset is still committed.

reason is separate from detail because one is a metric tag and the other is not. ADR §7 specifies consume.records.poison{reason}; tagging that counter with a parser message would mint a Prometheus timeseries per distinct malformed record, which is the unbounded-cardinality failure §7.1 calls the most likely way to take Prometheus down. reason is drawn from this closed set of four; detail goes to the log line and the DLQ payload, where cardinality costs nothing.

Producer and consumer wiring: the only place in the build that turns an com.worxbend.kernel.event.Envelope into a ProducerRecord, or a ConsumerRecord back into an envelope.

The asymmetry between the two directions is the point.

Producing may throw. Serializer.serialize runs on the caller's thread inside KafkaProducer.send, the exception surfaces as a failed Future at the call site that built the event, and the event is nobody else's problem yet. Failing loudly there is right: an envelope this build cannot encode is a bug in wolfram's validation, and dropping it quietly would lose an event a client was told had been accepted.

Consuming may never throw. ADR §4.3 gives the failure in full: a throwing deserializer throws inside KafkaConsumer.poll, before the Pekko connector ever sees the record, so the stream dies with the offset uncommitted and the restart replays the same record forever. Every decode here therefore yields Either[DecodeFailure, Envelope], and a poison record becomes a DeadLetter whose offset is still committed.

That is also why io.cloudevents.kafka.CloudEventDeserializer is not used and must not be introduced: it throws. envelopeDeserializer is safe in the same slot precisely because its value type is an Either — the hazard was never "deserializing in the consumer", it was "throwing in the consumer".

Kafka Headers seen as a string-keyed carrier: keys, get, put.

Structurally identical to com.worxbend.observability.TextCarrier[Headers], and named separately only because modules/eventing must not depend on modules/observability (ADR §3.3 fixes this module's dependencies at kernel, cloudevents, kafka-clients and opentelemetry-api). A module that sees both — wolfram, cobalt — binds the two together in one line:

given TextCarrier[Headers] with
 def keys(c: Headers) = KafkaHeaderCarrier.keys(c)
 def get(c: Headers, k: String) = KafkaHeaderCarrier.get(c, k)
 def put(c: Headers, k: String, v: String) = KafkaHeaderCarrier.put(c, k, v)

and from then on Tracing.inject / Tracing.spanFrom work over Kafka headers with no further plumbing.

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 stripping traceparent 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.

Where a record came from, in the coordinates that identify it uniquely and forever.

(topic, partition, offset) is the only identifier a record that failed to decode has — it has no CloudEvents id, because parsing the id is what failed. Everything about replay and about not double-writing the DLQ hangs off that triple.