Reliability & governance helpers

Beyond produce/consume, the Java core ships optional, zero-dependency helpers that mechanise the reliability and governance contracts the wire spec defines. None touch the frozen envelope (schema_version: 1); each is a tooling layer over the codec. They are the Java face of the cross-SDK spec — follow the linked spec page for the full contract.

Idempotency

com.babelqueue.idempotency.Idempotent.wrap(store, handler) makes a handler run at most once per meta.id, even on at-least-once redelivery.

import com.babelqueue.idempotency.Idempotent;
import com.babelqueue.idempotency.InMemoryStore;

Store store = new InMemoryStore(); // tests / single process
Handler guarded = Idempotent.wrap(store, handler);

The in-memory InMemoryStore is the reference; for a fleet, implement the Store interface (seen / remember / forget) over a shared backend (no persistent store ships in core yet — bring your own). See Idempotency and the store deep-dive.

Transactional outbox

The com.babelqueue.outbox package removes the producer dual write: persist the encoded envelope in the same DB transaction as your business row, then a relay publishes the durable rows verbatim.

import com.babelqueue.outbox.Outbox;
import com.babelqueue.outbox.OutboxRelay;
import com.babelqueue.outbox.InMemoryOutboxStore;

OutboxStore store = new InMemoryOutboxStore(); // production: a JDBC-backed OutboxStore
Outbox outbox = new Outbox(store);

// inside YOUR DB transaction, beside the business write — no commit of its own:
String id = outbox.write(envelope);            // encodes via EnvelopeCodec, calls store.save()

// later, a relay drains the durable rows to the broker:
OutboxRelayResult res = new OutboxRelay(transport, store).drain(0); // res.published() / res.failed()

OutboxStore is the four-method interface (save / fetchUnpublished / markPublished / markFailed); the relay forwards through OutboxTransport.publish(queue, body). The transaction boundary is yours. See Transactional Outbox.

DLQ redrive & replay-bypass

com.babelqueue.Redrive.redrive(transport, dlq, options) moves dead-lettered messages back onto a queue — dead_letter removed, attempts reset to 0, everything else preserved — with toQueue (sandbox), max, dryRun, select and bypass. The Replay guard skips effects that already fired on a deliberate replay; the replay flag rides a ThreadLocal scope (Replay.process(headers, …)), so Replay.isReplay() takes no argument:

import com.babelqueue.Replay;

Replay.process(headers, () -> {
    saveOrder(data);                                    // idempotent core — always runs
    Replay.bypassExternalEffects(() -> sendEmail(data)); // external effect — skipped on replay
});

Redrive with bypass(true) stamps the bq-replay-bypass marker through a HeaderPublisher transport. See DLQ redrive & replay-bypass.

com.babelqueue.gdpr.Gdpr.protect() / .unprotect() encrypt only the data leaves a schema marks x-gdpr-sensitive, in place. Cipher is caller-bound; AesGcmCipher (on the JDK’s javax.crypto) is the reference. Validate cleartext — protect after validation on produce, unprotect before validation on consume.

import com.babelqueue.gdpr.Gdpr;
import com.babelqueue.gdpr.AesGcmCipher;

Cipher cipher = new AesGcmCipher(key); // or implement com.babelqueue.gdpr.Cipher over a KMS

Gdpr.protect(data, schema, cipher);    // producer: encrypt marked leaves
try {
    Gdpr.unprotect(data, schema, cipher); // consumer: inverse
} catch (DecryptException e) {
    // wrong key / tampered → retry / dead-letter
}

com.babelqueue.schema.SensitivePaths.of(schema) exposes the marked leaves directly. See GDPR field encryption.

OpenTelemetry (`traceparent`)

com.babelqueue.otel.Tracing emits publish <urn> / process <urn> spans and, on a header-carrying transport, injects/extracts the W3C traceparent so a consumer span is a true child of the producer span — degrading to v0.1 trace_id correlation otherwise. Tracing.publish(…, HeaderSender) injects on produce and the Tracing.wrapHandler(tracer, handler, Supplier<Map<String,String>>) overload extracts on consume. The Spring AMQP, SQS and Redis transports carry the header. opentelemetry-api is the core’s already-optional dependency, so no new dependency is added. See Observability.

Per-URN schema validation

The com.babelqueue.schema validator (producer guard + consumer wrap, via MapProvider) validates a message’s data against the JSON Schema registered for its URN. See Per-URN schema validation.