Apache Kafka transport

Last updated: October 13, 2026

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

Maven:

<dependency>
    <groupId>com.babelqueue</groupId>
    <artifactId>babelqueue-kafka</artifactId>
    <version>1.2.1</version>
</dependency>

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

Produce

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

import com.babelqueue.kafka.KafkaConsumer;   // the BabelQueue worker (not kafka-clients' KafkaConsumer)
import com.babelqueue.kafka.RetryTopics;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import java.time.Duration;
import java.util.List;
import java.util.Map;

Consumer<byte[], byte[]> consumer = new org.apache.kafka.clients.consumer.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"));

// The retry / DLQ producer — the worker republishes failed envelopes through it.
Producer<byte[], byte[]> producer = new KafkaProducer<>(Map.of(
    ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
    ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class,
    ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class));

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

KafkaConsumer worker = 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

Both kafka-clients and BabelQueue ship a class called KafkaConsumer, so import only one of them and fully qualify the other, as above.

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 never lost to a failed write. Every retry-tier republish and every <topic>.dlq write (a handler that has run out of tries, an unknown URN, a poison body) is awaited until the broker confirms it, and only then is the offset committed. If the write fails, the offset is not committed: the consumer seeks the partition back to that record, reports the error to onError, and skips the rest of the batch, so the record is polled again and the handler may run again. Keep handlers idempotent.

The retry-tier republish and every <topic>.dlq write (the annotated envelope, the wrapped poison record and the raw schema-version rejection) carry the consumed 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 a failing record first (the retry republish or the <topic>.dlq write, then the commit) and calls onError afterwards, even when settling itself failed. An exception thrown by the callback is swallowed, so it cannot skip the DLQ write or the commit, and the poll loop keeps running. Handle errors inside the callback rather than rethrowing from it.

An undecodable (poison) body is never retried and never reaches 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 (not JSON, a scalar or array, trailing tokens, an empty value) becomes {"raw": "<body>"} (invalid UTF-8 is stored with U+FFFD substitutions), and a dead_letter block with reason: "poison", error, exception, original_queue, attempts (from the bq-attempts header) and lang: "java" is attached. The key is kept, and the offset is committed only after the DLQ write is confirmed (otherwise the partition is rewound).

A body is poison when, and only when, the Go reference’s typed decode would reject it: malformed JSON (leading zeros, NaN, trailing tokens, a BOM, a second object) or a non-object value; data, meta or dead_letter not an object; job, trace_id or a string member of meta / dead_letter not a string; meta.schema_version, meta.created_at, attempts or dead_letter.failed_at|attempts not an integer that fits int64 ("2026", a fraction or 1e2 is poison); or a number in data that is not a finite double (1e400, carried verbatim under raw). 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 on the unknown-URN path); 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: any RuntimeException or StackOverflowError from the scan or the codec is poison too. The nesting depth the recursive scanner carries is bounded by the thread’s stack (roughly 3000 to 5000 levels, depending on its size), where Go allows 10000; deeper bodies are poison. 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, with its original headers, is written to <topic>.dlq and the offset is then committed. If the DLQ write fails, the offset is not committed: the consumer seeks back to the record so it is polled again. With no DLQ configured the record is committed and dropped. onError is called with a null envelope. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation.

DLQ redrive

KafkaRedriveTransport adapts a Kafka Consumer subscribed to the DLQ topic (auto-commit off) and a Producer to the core’s Redrive.Transport, so Redrive.redrive (ADR-0026) can replay <topic>.dlq back to its work topic:

var transport = new KafkaRedriveTransport(dlqConsumer, producer);
Redrive.Result result = Redrive.redrive(transport, "orders.dlq", Redrive.Options.all());

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, 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, and stamps a fresh timestamp. The DLQ offset is committed only after the republish is acknowledged. 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 record. KafkaRedriveTransport therefore implements Redrive.Rewinder: whenever Redrive.redrive stops on a failed message (a failed restore write, a failed redrive write, a throwing select, a failed ack), the consumer seeks back to the first unacknowledged record of every partition it handed out in the batch (Redrive.redrive pops the whole batch before processing it). A failing seek does not stop the others and is reported together with the original error. An acknowledged record is never rewound: after a failed redrive write whose restore succeeded, only the rest of the batch is.

Contract mapping (§6)

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.