BabelQueue — Apache Kafka (Java)

com.babelqueue:babelqueue-kafka — an Apache Kafka transport for BabelQueue, built on the official kafka-clients and the framework-agnostic babelqueue-core.

A canonical-envelope publisher and a URN-routed, process-then-commit consumer, so a Kafka-based Java service speaks the same wire contract (envelope shape, URN identity, trace propagation) as the .NET, Python, Go and Node SDKs. Implements §6 of the broker-bindings contract.

Kafka has no native per-message ack, delayed delivery, dead-letter queue, or delivery counter — this transport absorbs all four in the binding layer (the envelope stays schema_version: 1):

• the envelope JSON is the record value; the contract fields are mirrored onto bq- headers (so a consumer routes on bq-job without decoding the body) and the record timestamp mirrors meta.created_at;
• bq-attempts is the authoritative retry counter (the body's attempts is the fallback for non-BabelQueue producers);
• consume is process-then-commit (manual commit, at-least-once);
• retry and delay use SDK-owned tiered retry topics <topic>.retry.<n>; a delay or release with no retry topics configured raises rather than silently dropping;
• terminal failures go to an opt-in <topic>.dlq topic carrying the canonical envelope plus the additive dead_letter block.

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

It pulls babelqueue-core and org.apache.kafka:kafka-clients transitively.

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", 1042));
}

publish(urn, data) returns the message meta.id; overloads add a traceId and a relative Duration delay. Delays require a retry topology (KafkaPublisher.create(producer, retryTopics)); on a plain publisher a delay raises BabelQueueException.

Map<String, Object> cfg = Map.of(
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
    ConsumerConfig.GROUP_ID_CONFIG, "orders-workers",
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false,            // manual commit is required
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class,
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);

var consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<byte[], byte[]>(cfg);
consumer.subscribe(List.of("orders"));

RetryTopics retry = RetryTopics.forTopic("orders")
    .tier(Duration.ofSeconds(5)).tier(Duration.ofMinutes(1)).build(); // .retry.1, .retry.2 + orders.dlq

KafkaConsumer worker = KafkaConsumer.builder(consumer)
    .producer(producer)            // republishes retry/DLQ records
    .retryTopics(retry)
    .maxTries(3)
    .handler("urn:babel:orders:created", (env, record) -> {
        // env.data(), env.traceId(), env.attempts() ...
    })
    .onError((err, env, record) -> err.printStackTrace())
    .build();

worker.run(() -> true); // poll → process → commit, until you stop it

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, so it never decodes a record it cannot handle. Unknown-URN strategy is one of fail / delete / release / dead_letter.

Schema-version gate (§6.7). Before decoding the value the consumer reads the bq-schema-version header (UTF-8). Absent or blank (empty / ASCII whitespace only) → normal decode. Any other value that is not exactly 1 (no trimming — "2", "x", " 1") → the record is not decoded and the handler is not called: onError is notified with a null envelope, the raw record is forwarded to <topic>.dlq (if configured) and the offset is committed.

Retry records are produced with bq-delay set to the tier delay. KafkaRetryReinjector puts them back on the work topic once the tier has elapsed (§6.4):

var retryConsumer = new org.apache.kafka.clients.consumer.KafkaConsumer<byte[], byte[]>(cfg); // own group.id
KafkaRetryReinjector reinjector = KafkaRetryReinjector.builder(retryConsumer, producer, retry).build();
reinjector.subscribe();            // every <topic>.retry.<n>
reinjector.run(() -> true);        // wait → forward → commit

• Due time is record.timestamp + tier_ms, clamped to [now, now + tier_ms]: a record older than its tier is forwarded immediately, and a skewed (future) timestamp never waits more than one tier.
• The wait is partition pause/resume; the poll loop is never blocked.
• Value and key are written to bq-original-topic byte-for-byte; the body is not decoded and attempts is not touched. bq-delay and bq-original-topic are dropped, every other header (e.g. traceparent) is kept.
• The offset is committed only after the broker acknowledged the write (at-least-once; the work consumer dedupes on meta.id).
• Opt-in default tier ladder: RetryTopics.forTopic("orders").recommendedTiers() = 5 s, 30 s, 5 min, 30 min. The failed attempt n goes to <topic>.retry.<min(attempts_before, tiers - 1) + 1>.

Operator note. Set message.timestamp.type=LogAppendTime on the retry topics (SHOULD): the wait is measured from the record timestamp, and the broker's append time is immune to producer clock skew.

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 preserved byte-for-byte on the redrive write and on the restore write back to the DLQ (null stays null, empty stays empty — contract §6.5). The DLQ offset is committed only after the re-publish is acknowledged. A restore write re-writes the DLQ record verbatim (original value bytes, headers and key), so a record that is not valid UTF-8 is never altered. Needs babelqueue-core 1.10.0 or newer (Redrive.KeyedRepublisher, Redrive.Rewinder).

All header values are UTF-8 byte strings (integers as decimal strings, e.g. "1"). The envelope is unchanged (schema_version stays 1); Apache Kafka is purely additive.

mvn verify

The Kafka Consumer is mocked with Mockito and the producer with the official MockProducer — no Kafka, no network. JUnit 5, JaCoCo ≥90% line coverage.

MIT