README.md

BabelQueue — Amazon SQS (Java)

com.babelqueue:babelqueue-sqs — an Amazon SQS transport for BabelQueue, built on the AWS SDK for Java v2 and the framework-agnostic babelqueue-core.

A canonical-envelope publisher and a URN-routed consumer, so an SQS-based Java service speaks the same wire contract (envelope shape, URN identity, trace propagation) as the PHP/Laravel, Python, Go, Node and .NET SDKs. Implements §3 of the broker-bindings contract.

Install (Maven)

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

It pulls babelqueue-core and software.amazon.awssdk:sqs transitively.

Use

SqsClient sqs = SqsClient.create(); // your AWS config / credentials chain
String url = "https://sqs.eu-central-1.amazonaws.com/123456789012/orders";

// produce
String id = SqsPublisher.create(sqs, url)
    .publish("urn:babel:orders:created", Map.of("order_id", 1042));

// consume
SqsConsumer consumer = SqsConsumer.builder(sqs, url)
    .handler("urn:babel:orders:created", (env, message) -> {
        // env.data(), env.traceId(), env.attempts() ...
    })
    .onError((err, env, message) -> log.warn("bad message", err))
    .build();
consumer.run(); // long-polls until the thread is interrupted

FIFO: SqsPublisher.builder(sqs, url).fifo(true).build() (the URL must end in .fifo). For LocalStack/ElasticMQ, point the SqsClient's endpoint there.

Contract mapping (§3)

Envelope	SQS
body	MessageBody (byte-identical across SDKs)
job (URN)	MessageAttributes.bq-job
trace_id	MessageAttributes.bq-trace-id
meta.id	MessageAttributes.bq-message-id
meta.schema_version	MessageAttributes.bq-schema-version (Number)
meta.lang	MessageAttributes.bq-source-lang
meta.created_at	MessageAttributes.bq-created-at (Number, ms)
attempts	reconciled to ApproximateReceiveCount − 1 on receive
reserve / ack	visibility timeout → DeleteMessage

Retry is SQS-native: a throwing handler's message is never deleted — it is released with ChangeMessageVisibility(ReceiptHandle, VisibilityTimeout = releaseDelaySeconds) and SQS redelivers it (at-least-once). The default delay is 0 (redeliver now); out-of-range values are clamped to 0–43200 s. A failed DeleteMessage after a successful handler is reported as SqsDeleteException and is not released (the message returns after its visibility timeout). The poll loop never stops on a bad message, a failed release or a failed delete — observe via onError / onUnknownUrn. The envelope is unchanged (schema_version stays 1); SQS is purely additive.

Schema-version gate (§3.7). Before decoding the body the consumer reads the bq-schema-version attribute. Missing or blank (empty / ASCII whitespace only) → normal decode. Any other value that is not exactly 1 (no trimming — "2", "x", " 1") → the body is not decoded, the handler is not called, onError is notified with a null envelope, and the message is left undeleted so SQS redrives it to the DLQ.

SqsConsumer.builder(sqs, url)
    .handler("urn:babel:orders:created", handler)
    .releaseDelaySeconds(30)                          // backoff between attempts (default 0)
    .unknownUrnStrategy(UnknownUrnStrategy.RELEASE)   // fail | delete | release | dead_letter
    .unknownUrnReleaseDelaySeconds(60)                // default 0
    .build();

Poison messages. With the default 0 delay, a message that always fails is redelivered on every receive. Give the queue a RedrivePolicy (deadLetterTargetArn + maxReceiveCount) so SQS moves it to a native DLQ after N receives, and set a non-zero releaseDelaySeconds when the handler depends on something that can be briefly unavailable — otherwise a short outage can use up maxReceiveCount in under a second.

Trace propagation (OpenTelemetry traceparent, ADR-0028)

The optional core com.babelqueue.otel module can carry a W3C traceparent so a consumer span becomes a true child of the producer span — propagated out of band on the SQS MessageAttributes channel, beside the contract bq-* attributes (a contract attribute always wins a key collision; bounded by the SQS 10-attribute cap), never inside the frozen envelope (GR-1).

// produce: HeaderSender -> SqsPublisher.publishWithHeaders
SqsPublisher publisher = SqsPublisher.create(sqs, url);
Tracing.publish(tracer, "urn:babel:orders:created", Map.of("order_id", 1042), "orders",
    (envelope, headers) -> publisher.publishWithHeaders(envelope, headers));

// consume: surface the delivered attributes for wrapHandler's Supplier
SqsConsumer.builder(sqs, url)
    .handler("urn:babel:orders:created", (env, message) ->
        Tracing.wrapHandler(tracer, h, () -> SqsHeaders.of(message)).handle(env))
    .build();

A header-less publish(...) is byte-identical to before; with no traceparent the consumer falls back to the v0.1 trace_id-derived parent. Requires babelqueue-core ≥ 1.5.0. No OpenTelemetry dependency is needed unless you opt in — the seam is a plain Map<String,String>.

Build & test

mvn verify

Unit tests run against a fake SqsClient (no AWS, no network). AWS SDK v2's SqsClient is an interface, so the fake overrides only the three operations used.

License

MIT