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

# Apache Kafka transport

`com.babelqueue:babelqueue-kafka` is an Apache Kafka transport on the Java 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

Maven:

```xml
<dependency>
    <groupId>com.babelqueue</groupId>
    <artifactId>babelqueue-kafka</artifactId>
    <version>1.0.0</version>
</dependency>
```

Requirements: **Java 17+**. It pulls `babelqueue-core` and `org.apache.kafka:kafka-clients`
transitively. You supply the Kafka `Producer` / `Consumer`.

## Produce

```java
import com.babelqueue.kafka.KafkaPublisher;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import java.util.Map;

Map<String, Object> cfg = Map.of(
    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class,
    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);

try (Producer<byte[], byte[]> producer = new KafkaProducer<>(cfg)) {
    String id = KafkaPublisher.create(producer, "orders")
        .publish("urn:babel:orders:created", Map.of("order_id", 1042L));
}
```

`publish(urn, data)` returns the message `meta.id`; overloads add a `traceId` and a relative
`Duration delay` (delays require a `RetryTopics` topology, else they raise).

## Consume

```java
import com.babelqueue.kafka.*;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import java.time.Duration;
import java.util.List;

Consumer<byte[], byte[]> consumer = new KafkaConsumer<>(Map.of(
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
    ConsumerConfig.GROUP_ID_CONFIG, "orders-workers",
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,      // manual commit (process-then-commit)
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class,
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class));
consumer.subscribe(List.of("orders"));

RetryTopics retry = RetryTopics.forTopic("orders")
    .tier(Duration.ofSeconds(5)).tier(Duration.ofMinutes(1)).build();

com.babelqueue.kafka.KafkaConsumer worker = com.babelqueue.kafka.KafkaConsumer.builder(consumer)
    .producer(producer)
    .retryTopics(retry)
    .maxTries(3)
    .handler("urn:babel:orders:created", (envelope, record) -> {
        // envelope.data(), envelope.traceId(), envelope.attempts() ...
    })
    .build();

worker.run(() -> true); // poll → 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 | poll → process → **commit offset** (manual) |
| retry / delay | republish to `<topic>.retry.<n>` (`bq-attempts + 1`) |
| dead-letter | `<topic>.dlq` + `dead_letter` block |

The Kafka `Consumer` is mocked with Mockito and the producer with the official `MockProducer`
— no Kafka, no network. The envelope is unchanged (`schema_version` stays `1`); Apache Kafka
is purely additive.