Apache Kafka transport

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

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.*;
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)

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.