---
title: "Apache Kafka transport"
description: "The BabelQueue.Kafka transport — a canonical-envelope publisher and a URN-routed, process-then-commit consumer over Confluent.Kafka, on the framework-agnostic core."
source: https://babelqueue.com/docs/babelqueue-dotnet/1.x/apache-kafka/
updated: 2026-06-14T00:00:00.000Z
---

# Apache Kafka transport

`BabelQueue.Kafka` is an Apache Kafka transport on the .NET core. It sends the
[canonical envelope](/docs/spec/1.x/envelope/) as the record value with the
[§6 header projection](/docs/spec/1.x/broker-bindings/#apache-kafka), 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

```bash
dotnet add package BabelQueue.Kafka
```

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

## Produce

```csharp
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

```csharp
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)

| 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` |
| `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.