BabelQueue — Apache Kafka (.NET)

BabelQueue.Kafka — an Apache Kafka transport for BabelQueue, built on Confluent.Kafka and the framework-agnostic BabelQueue.Core.

A canonical-envelope publisher and a URN-routed, process-then-commit consumer, so a Kafka-based .NET service speaks the same wire contract (envelope shape, URN identity, trace propagation) as the Java, Python, Go and Node SDKs. Implements §6 of the broker-bindings contract.

Kafka has no native per-message ack, delayed delivery, dead-letter queue, or delivery counter — this transport absorbs all four in the binding layer (the envelope stays schema_version: 1): the record value is the envelope JSON, the contract fields are mirrored onto bq- headers (route on bq-job without decoding the body), the record timestamp mirrors meta.created_at, bq-attempts is the authoritative retry counter, consume is process-then-commit (manual commit), retry/delay use SDK-owned tiered retry topics <topic>.retry.<n>, and terminal failures go to an opt-in <topic>.dlq.

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, or a delay (TimeSpan) — delays require a retry topology (KafkaPublisher.Create(producer, retryTopics)) and route 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
    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. An undecodable / poison body (invalid UTF-8, malformed JSON, not a JSON object, or any typed member the strict decode rejects such as meta.created_at: "2026", with member names matched case-sensitively as the contract spells them, so a re-cased name is an unknown member, not poison; a well-typed but non-conformant envelope is not poison and is routed by URN) is dead-lettered wrapped with dead_letter.reason = "poison" (Go-style wrapping, except that invalid UTF-8 or a lone UTF-16 surrogate escape is poison here; any exception from the decode is also poison, never a stuck partition; a body that is one JSON object keeps its keys, anything else becomes {"raw": "<body>"}), key preserved. Before decoding, the consumer reads the bq-schema-version header (UTF-8): absent or blank decodes as usual, any other value must be exactly "1", otherwise the raw record is forwarded to <topic>.dlq and committed without being decoded (with no DLQ topic configured the forward is a no-op, so the record is just committed — it degrades to delete). If the DLQ write itself fails, the record is not committed: the consumer seeks back to its offset (so a later record's cumulative commit cannot skip it), reports via OnError (after the seek; a throwing OnError cannot bypass it), backs off for one PollTimeout and the loop keeps running. A header with a null value counts as absent. The consumer routes on the bq-job header. Unknown-URN strategy is one of fail / delete / release / dead_letter.

Redrive (DLQ replay)

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 redriven record carries the DLQ record's key byte-for-byte (null stays null, empty stays empty, none is synthesised — contract §6.5), on both the redrive write to the work topic and every restore write back to the DLQ (dry run, unselected, undecodable body, failed redrive write). A restore re-writes the DLQ record verbatim (original value bytes, headers and key, fresh timestamp); a record that is not valid UTF-8 is restored byte-for-byte and committed only after that write succeeded. A redrive to the work topic re-projects the bq- headers from the reset envelope; the bq-replay-bypass header is not propagated. Offsets are committed only after the re-publish (at-least-once). Requires BabelQueue.Core 1.10.0 or newer (Redrive.IKeyedRepublisher, Redrive.IRewinder).

Contract mapping (§6)

Envelope	Apache Kafka
body	record value (byte-identical across SDKs)
job (URN)	header bq-job (consumer routes on this)
trace_id	header bq-trace-id
meta.id	header bq-message-id
meta.schema_version	header bq-schema-version ("1")
meta.lang	header bq-source-lang
meta.created_at	record Timestamp (Unix ms)
attempts	header bq-attempts (authoritative; body is the fallback)
reserve / ack	consume → process → commit offset (manual)
retry / delay	republish to <topic>.retry.<n> (bq-attempts + 1)
dead-letter	<topic>.dlq + dead_letter block

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.

Build & test

dotnet test

xUnit; analyzers run as errors; ≥90% line coverage enforced.

License

MIT