Apache Kafka transport

Last updated: October 13, 2026

BabelQueue.Kafka is an Apache Kafka transport on the .NET core. It sends the canonical envelope as the record value with the §6 header projection, and consumes by routing each record to a handler by URN — so a record it produces is consumed by any other BabelQueue SDK, and vice-versa. Kafka has no native ack/delay/DLQ/delivery-counter, so the binding absorbs all four in the transport.

Install

dotnet add package BabelQueue.Kafka

Requirements: .NET 8. It pulls BabelQueue.Core and Confluent.Kafka transitively.

Produce

using Confluent.Kafka;
using BabelQueue.Kafka;

using var producer = new ProducerBuilder<byte[], byte[]>(
    new ProducerConfig { BootstrapServers = "localhost:9092" }).Build();

var id = await KafkaPublisher.Create(producer, "orders")
    .PublishAsync("urn:babel:orders:created", new Dictionary<string, object?> { ["order_id"] = 1042 });

PublishAsync returns the message meta.id; pass a traceId to continue a trace. A delay (TimeSpan) requires a retry topology (KafkaPublisher.Create(producer, retryTopics)) and routes to the matching tier; on a plain publisher a delay raises BabelQueueException.

Consume

using var consumer = new ConsumerBuilder<byte[], byte[]>(new ConsumerConfig
{
    BootstrapServers = "localhost:9092",
    GroupId = "orders-workers",
    EnableAutoCommit = false,            // manual commit is required (process-then-commit)
    AutoOffsetReset = AutoOffsetReset.Earliest,
}).Build();
consumer.Subscribe("orders");

var retry = RetryTopics.ForTopic("orders")
    .Tier(TimeSpan.FromSeconds(5)).Tier(TimeSpan.FromMinutes(1)).Build(); // .retry.1/.2 + orders.dlq

var worker = new KafkaConsumer(consumer, new Dictionary<string, BabelHandler>
{
    ["urn:babel:orders:created"] = (env, result, ct) =>
    {
        // env.Data, env.TraceId, env.Attempts ...
        return Task.CompletedTask;
    },
}, new KafkaConsumerOptions { Producer = producer, RetryTopics = retry, MaxTries = 3 });

await worker.RunAsync(cancellationToken); // consume → process → commit

A throwing handler republishes the envelope to the next <topic>.retry.<n> tier with bq-attempts + 1, then commits; once MaxTries is reached it goes to <topic>.dlq with a dead_letter block. The consumer routes on the bq-job header. Unknown-URN strategy is one of fail / delete / release / dead_letter.

A record is not lost when a settlement write fails. If the retry-tier republish or the <topic>.dlq write (a handler that has run out of tries, an unknown URN, a poison body) throws, or is cancelled, the offset is not committed: the consumer seeks the partition back to that record, so the next poll returns it again (and the handler may run again), and the error is rethrown. If the seek itself fails, a BabelQueueException wraps both errors. This relies on EnableAutoCommit=false, which the consumer already requires: with auto-commit on, the SDK cannot protect the record.

With no retry tier and no DLQ configured, a throwing handler raises a BabelQueueException, the consumer seeks the partition back, and the record is redelivered on every poll. The partition does not advance past that record until the handler succeeds or you configure a retry tier or DLQ.

The retry-tier republish and every <topic>.dlq write (the annotated dead_letter record as well as the raw record written for a rejected schema version, and the wrapped poison record) carry the source record’s key unchanged: a null key stays null, an empty key stays empty. The key has no contract meaning and the transport never makes one up. Partition equality across SDKs is not guaranteed, because client partitioners differ.

OnError never blocks settlement. The consumer settles first (the retry republish or the <topic>.dlq write, then the commit) and calls OnError afterwards. On a handler failure or a poison body, if settling fails the consumer seeks back, reports, and rethrows. The schema-version gate and invalid UTF-8 paths below behave as described there. An exception thrown by the callback is swallowed, so it cannot skip the DLQ write or the commit, and the consumer keeps running.

A poison body is never retried and never handed to a handler. It is dead-lettered wrapped, like the Go reference (§6.5): a body that is exactly one JSON object keeps its keys, anything else becomes {"raw": "<body>"}, and a dead_letter block with reason: "poison", error, exception, original_queue, attempts (from the bq-attempts header) and lang is attached. The record key is kept, the bq-* headers are projected from the body, and the offset is committed once the DLQ write is confirmed; then OnError is called. If the DLQ write fails the offset is not committed, the consumer seeks back to the record, calls OnError and rethrows the write error; with no DLQ topic configured the record is committed and dropped.

A body is poison when the Go reference’s typed decode would reject it (plus the two differences below): it is not valid JSON (leading zeros, NaN, trailing tokens, a BOM, an empty body) or not a JSON object; data, meta or dead_letter is not an object; job, trace_id or a string member of meta / dead_letter is not a string; meta.schema_version, meta.created_at, attempts or dead_letter.failed_at|attempts is not an integer that fits int64 ("2026", a fraction or 1e2 is poison); or a number in data is not a finite double (1e400). Member names are matched exactly as spelled (JOB or Meta is an unknown member, kept as it is), and every duplicate key is type-checked. A bare JSON null is not poison (an empty envelope, which takes the unknown-URN path), and a well-typed but non-conformant envelope (a missing data, a blank trace_id, a schema_version other than 1) is routed by its URN like any other record. No decode failure can wedge a partition: whatever the codec throws on a body the scan accepted is poison too.

Two differences from Go are intentional. A body that is not valid UTF-8 is poison here, where Go substitutes U+FFFD and accepts it. And a body nested more than 64 levels (counting the root object; the core codec’s System.Text.Json depth limit) is poison, where Go decodes much deeper. It is parked verbatim with its keys kept. A 1.0 or 1e2 float literal is re-formatted where a body is re-encoded; the original text is kept only under raw. On a poison record, dead_letter.attempts is the bq-attempts header (Go writes 0) and original_queue is the bq-original-topic header when present.

Before decoding, the consumer version-gates on the bq-schema-version record header (§6.7). A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A header present with a null value counts as missing too. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body: the raw record is written to <topic>.dlq and the offset is then committed (with no DLQ topic configured the record is committed and dropped). If the DLQ write fails, the offset is not committed and the consumer seeks back to the record so it is polled again. OnError is called with an empty envelope. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation.

DLQ redrive

KafkaRedriveTransport plugs a <topic>.dlq topic into the core’s Redrive.RedriveAsync (ADR-0026):

// consumer: subscribed to "orders.dlq", EnableAutoCommit = false
var transport = new KafkaRedriveTransport(consumer, producer);
var result = await Redrive.RedriveAsync(transport, "orders.dlq", new Redrive.Options());

The DLQ record’s key is kept byte for byte on the write back to the work topic and on every restore write back to the DLQ (dry run, unselected, undecodable body, failed redrive write): a null key stays null, an empty key stays empty. A restore re-writes the DLQ record verbatim, with its original value bytes, headers and key, so a record that is not valid UTF-8 is never altered. A restore stamps a fresh timestamp, so the restored copy does not age out of the DLQ early. Offsets are committed only after the republish succeeds. The transport does not propagate the bq-replay-bypass header.

Kafka commits are cumulative, so a failed run must not leave the group past an unsettled DLQ record. KafkaRedriveTransport therefore implements the core’s Redrive.IRewinder (BabelQueue.Core 1.10.0 or later). When a failed restore write, a failed redrive write, a failed commit or a throwing Select callback ends the run, Redrive.RedriveAsync calls RewindAsync, which seeks every partition that still has a handed-out, uncommitted record back to its lowest such offset, so the rest of the drained batch is redelivered on the next poll. Records already committed (a restored record included) are never sought back to, so nothing is duplicated. If the seek fails too, both errors are reported together (AggregateException).

Contract mapping (§6)

The IProducer / IConsumer interfaces are mockable, so the unit tests use Moq — no Kafka, no network. The envelope is unchanged (schema_version stays 1); Apache Kafka is purely additive.