Apache Kafka transport

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.

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.