diff --git a/CHANGELOG.md b/CHANGELOG.md index 30864d9..85ce61d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,23 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +## [1.5.0] - 2026-11-03 + +### Added + +- `SqsConsumer.Builder.deadLetterObserver(DeadLetterObserver)` (ADR-0039 section 5.1, opt-in): the observer is called + once for every record this consumer **sends** to the configured `deadLetterQueueUrl` - a version-gated raw + body (empty `reason`), a poison body (`poison`) and an unknown URN under `dead_letter` (`unknown_urn`) - after the + send is confirmed and before the source message is deleted. The event carries the source queue (the + envelope's `meta.queue` for an unknown URN, else the consumer's queue name; a poison or version-gated delivery + always reports the consumer's queue), the DLQ queue name and the `dead_letter.reason`. The call is at-least-once: + if the later `DeleteMessage` fails, the message is redelivered and sent again, and the observer is called again + (a FIFO DLQ deduplicates the record, not the call). It is not called for a failed send, for a message left to the queue's + native `RedrivePolicy` (no DLQ queue, or retries exhausted - the SDK does not dead-letter those), or for a + `delete`/degraded drop. An observer that throws is logged and swallowed; the delete, release and DLQ outcome are + unchanged. Unset (the default) behaviour is byte-for-byte unchanged. Wire format unchanged. +- Requires a `babelqueue-core` that carries `DeadLetterObserver` / `DeadLetterEvent` (core 1.15.0, the version this release pins). + ## [1.4.1] - 2026-10-26 ### Changed diff --git a/README.md b/README.md index c7aab85..4dfcf9e 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ as the PHP/Laravel, Python, Go, Node and .NET SDKs. Implements com.babelqueue babelqueue-sqs - 1.4.1 + 1.5.0 ``` diff --git a/pom.xml b/pom.xml index 44e9bea..944a899 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ com.babelqueue babelqueue-sqs - 1.4.1 + 1.5.0 jar BabelQueue Amazon SQS @@ -49,7 +49,7 @@ 17 UTF-8 UTF-8 - 1.14.0 + 1.15.0 2.25.0 1.45.0 6.1.3 diff --git a/src/main/java/com/babelqueue/sqs/SqsConsumer.java b/src/main/java/com/babelqueue/sqs/SqsConsumer.java index d6bb360..2621fb6 100644 --- a/src/main/java/com/babelqueue/sqs/SqsConsumer.java +++ b/src/main/java/com/babelqueue/sqs/SqsConsumer.java @@ -1,6 +1,8 @@ package com.babelqueue.sqs; import com.babelqueue.BabelQueueException; +import com.babelqueue.DeadLetterEvent; +import com.babelqueue.DeadLetterObserver; import com.babelqueue.DeadLetters; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; @@ -112,6 +114,7 @@ public interface UnknownUrnHandler { private final boolean retryJitterFull; private final DoubleSupplier random; private final int unknownUrnReleaseDelaySeconds; + private final DeadLetterObserver deadLetterObserver; /** The SQS maximum for {@code VisibilityTimeout}: 12 hours, in seconds. */ public static final int MAX_VISIBILITY_TIMEOUT_SECONDS = 43_200; @@ -136,6 +139,7 @@ private SqsConsumer(Builder builder) { this.retryJitterFull = builder.retryJitterFull; this.random = builder.random != null ? builder.random : () -> ThreadLocalRandom.current().nextDouble(); this.unknownUrnReleaseDelaySeconds = builder.unknownUrnReleaseDelaySeconds; + this.deadLetterObserver = builder.deadLetterObserver; } public static Builder builder(SqsClient client, String queueUrl) { @@ -231,7 +235,7 @@ private void settleVersionGated(Message message, String declaredVersion) { // Rule 4: the raw body, byte for byte, then delete. Rule 5: a failed send leaves it unsettled. String body = message.body() == null ? "" : message.body(); if (sendToDeadLetterQueue(message, body, message.hasMessageAttributes() - ? message.messageAttributes() : Map.of(), null)) { + ? message.messageAttributes() : Map.of(), null, "", SqsQueues.nameFromUrl(queueUrl))) { delete(message, null); } } @@ -260,7 +264,7 @@ record = poisonRecord(body.getBytes(StandardCharsets.UTF_8), SqsQueues.nameFromU return; } if (sendToDeadLetterQueue(message, record, message.hasMessageAttributes() - ? message.messageAttributes() : Map.of(), partial)) { + ? message.messageAttributes() : Map.of(), partial, "poison", SqsQueues.nameFromUrl(queueUrl))) { delete(message, partial); } } @@ -291,9 +295,16 @@ private static int receiveAttempts(Message message) { * so the caller leaves the message unsettled and SQS redelivers it once the visibility timeout expires. A FIFO * DLQ ({@code .fifo}) gets the consumer's queue name as the message group and the SQS message id (else a * digest of the body) as the deduplication id, so a send repeated after a lost delete is not duplicated. + * + *

After each confirmed send the optional {@link DeadLetterObserver} is called once (ADR-0039 section 5.1) with + * {@code reason} ({@code ""} for a raw version-gated body) and {@code eventQueue}; a failed send does not call + * it. The observer is isolated through {@link DeadLetterObserver#notifySafely}, so it never changes the settle + * outcome. If the later {@code DeleteMessage} fails the message is redelivered and sent again, so the observer + * is called again (at-least-once; a FIFO DLQ deduplicates the record but not this call). */ private boolean sendToDeadLetterQueue( - Message message, String body, Map attributes, Envelope envelope) { + Message message, String body, Map attributes, Envelope envelope, + String reason, String eventQueue) { try { SendMessageRequest.Builder request = SendMessageRequest.builder() .queueUrl(deadLetterQueueUrl) @@ -308,11 +319,28 @@ private boolean sendToDeadLetterQueue( : sha256Hex(body)); } client.sendMessage(request.build()); - return true; } catch (RuntimeException sendError) { report(sendError, envelope, message); return false; } + if (deadLetterObserver != null) { + DeadLetterObserver.notifySafely(deadLetterObserver, new DeadLetterEvent( + eventQueue, SqsQueues.nameFromUrl(deadLetterQueueUrl), reason)); + } + return true; + } + + /** + * The source queue of a dead-letter event for a decoded, accepted envelope: its {@code meta.queue}, else this + * consumer's queue (ADR-0039 section 5.1). A poison or version-gated delivery has no accepted envelope and + * always reports the consumer's queue. + */ + private String sourceQueue(Envelope envelope) { + if (envelope != null && envelope.meta() != null + && !EnvelopeCodec.isBlankAscii(envelope.meta().queue())) { + return envelope.meta().queue(); + } + return SqsQueues.nameFromUrl(queueUrl); } private static String sha256Hex(String value) { @@ -380,7 +408,8 @@ private void deadLetterUnknownUrn(Message message, Envelope envelope) { try { dead = DeadLetters.annotate(envelope, "unknown_urn", SqsQueues.nameFromUrl(queueUrl)); body = EnvelopeCodec.encode(dead); - if (sendToDeadLetterQueue(message, body, SqsAttributes.project(dead), envelope)) { + if (sendToDeadLetterQueue(message, body, SqsAttributes.project(dead), envelope, "unknown_urn", + sourceQueue(envelope))) { delete(message, envelope); } } catch (RuntimeException error) { @@ -510,6 +539,7 @@ public static final class Builder { private boolean retryJitterFull; private DoubleSupplier random; private int unknownUrnReleaseDelaySeconds; + private DeadLetterObserver deadLetterObserver; private Builder(SqsClient client, String queueUrl) { this.client = Objects.requireNonNull(client, "client"); @@ -664,6 +694,28 @@ public Builder unknownUrnReleaseDelaySeconds(int seconds) { return this; } + /** + * Optional observer of the dead-letters this consumer itself writes (ADR-0039 section 5.1): called once + * per record sent to the {@link #deadLetterQueueUrl DLQ queue} (a version-gated raw body, a poison + * body, an unknown URN under {@code dead_letter}), after the send is confirmed and before the source message + * is deleted. The event carries the source queue (the envelope's {@code meta.queue} + * for an unknown URN, else this consumer's queue name; a poison or version-gated delivery always reports + * the consumer's queue), the DLQ queue name and the {@code dead_letter.reason} ({@code ""} for a raw + * version-gated body). It is at-least-once: if the later {@code DeleteMessage} fails the message is + * redelivered and sent again, and the observer is called again (a FIFO DLQ deduplicates the record, not + * the call). + * + *

It is not called when the send fails, when the message is left to the queue's native + * {@code RedrivePolicy} (no DLQ queue, or retries exhausted), or when it is deleted instead of dead-lettered. + * Anything the observer throws is logged and swallowed; the delete, release and DLQ outcome never change. + * Unset (the default): behaviour is unchanged. {@code com.babelqueue.otel.Metrics#deadLetterObserver} + * increments {@code babelqueue.dlq.messages}. + */ + public Builder deadLetterObserver(DeadLetterObserver observer) { + this.deadLetterObserver = observer; + return this; + } + /** Clamp a delay to the SQS {@code VisibilityTimeout} range, 0–43200 s (Go/Python parity). */ private static int clampDelay(int seconds) { return Math.max(0, Math.min(seconds, MAX_VISIBILITY_TIMEOUT_SECONDS)); diff --git a/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java b/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java index 27f51f2..e3b9ec5 100644 --- a/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java +++ b/src/test/java/com/babelqueue/sqs/SqsConsumerTest.java @@ -9,6 +9,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import com.babelqueue.BabelQueueException; +import com.babelqueue.DeadLetterEvent; import com.babelqueue.Envelope; import com.babelqueue.EnvelopeCodec; import com.babelqueue.UnknownUrnException; @@ -727,4 +728,221 @@ void blankDeadLetterQueueUrlIsRejected() { assertThrows(IllegalArgumentException.class, () -> SqsConsumer.builder(new FakeSqsClient(), URL).deadLetterQueueUrl(" ")); } + + // ---- dead-letter observer (ADR-0039 section 5.1) ----------------------------------------- + + private static final DeadLetterEvent POISON_EVENT = new DeadLetterEvent("orders", "orders.dlq", "poison"); + + @Test + void observerIsCalledOnceForAVersionGatedRawBodyWithAnEmptyReason() { + FakeSqsClient sqs = new FakeSqsClient(); + sendWithVersion(sqs, "not json", "2"); + sqs.sent.clear(); + List events = new ArrayList<>(); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "")), events); + assertEquals(1, sqs.sent.size()); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void observerIsCalledOnceForAPoisonBody() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "plain text, not JSON", 1); + List events = new ArrayList<>(); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(POISON_EVENT), events); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void observerIsCalledOnceForAnUnknownUrnUnderTheDeadLetterStrategy() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, validBody(), 1); + List events = new ArrayList<>(); + + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).deadLetterObserver(events::add).build().poll(); + + assertEquals(List.of(new DeadLetterEvent("orders", "orders.dlq", "unknown_urn")), events); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void unknownUrnEventReportsMetaQueueElsePoisonAndGateReportTheConsumerQueue() { + // meta.queue differs from the consumer's queue: the unknown-URN event carries meta.queue (ADR-0039 5.1). + FakeSqsClient billing = new FakeSqsClient(); + billing.seed(URL, EnvelopeCodec.encode( + EnvelopeCodec.make("urn:babel:orders:created", Map.of("order_id", 7), "billing", null)), 1); + List events = new ArrayList<>(); + SqsConsumer.builder(billing, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).deadLetterObserver(events::add).build().poll(); + assertEquals(List.of(new DeadLetterEvent("billing", "orders.dlq", "unknown_urn")), events); + + // A poison body whose JSON carries a meta.queue is still non-conformant: the consumer's queue is reported. + FakeSqsClient poison = new FakeSqsClient(); + poison.seed(URL, "{\"meta\":{\"queue\":\"billing\"}}", 1); + sendWithVersion(poison, "not json", "2"); + events.clear(); + consumerWithDlq(poison, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + consumerWithDlq(poison, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + assertEquals(2, events.size()); + events.forEach(e -> assertEquals("orders", e.queue())); + } + + @Test + void observerSeesTheDlqWriteBeforeTheSourceMessageIsDeleted() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + int[] deletedAtCall = {-1}; + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]) + .deadLetterObserver(e -> deletedAtCall[0] = sqs.deleted.size()).build().poll(); + + assertEquals(0, deletedAtCall[0]); + assertEquals(1, sqs.deleted.size()); + } + + @Test + void observerIsNotCalledWhenTheDlqWriteFails() { + List events = new ArrayList<>(); + + FakeSqsClient poison = new FakeSqsClient(); + poison.seed(URL, "{}", 1); + poison.sendError = new IllegalStateException("dlq down"); + consumerWithDlq(poison, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + + FakeSqsClient gated = new FakeSqsClient(); + sendWithVersion(gated, "not json", "2"); + gated.sendError = new IllegalStateException("dlq down"); + consumerWithDlq(gated, new ArrayList<>(), new int[1]).deadLetterObserver(events::add).build().poll(); + + FakeSqsClient unknown = new FakeSqsClient(); + unknown.seed(URL, validBody(), 1); + unknown.sendError = new IllegalStateException("dlq down"); + SqsConsumer.builder(unknown, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).deadLetterObserver(events::add).build().poll(); + + assertTrue(events.isEmpty()); + assertTrue(poison.deleted.isEmpty()); + assertTrue(gated.deleted.isEmpty()); + assertTrue(unknown.deleted.isEmpty()); + } + + @Test + void observerIsNotCalledWhenNothingIsWrittenToTheDlq() { + List events = new ArrayList<>(); + + // No DLQ queue: poison and version-gated bodies are left to the native RedrivePolicy. + FakeSqsClient native1 = new FakeSqsClient(); + native1.seed(URL, "{}", 1); + SqsConsumer.builder(native1, URL).deadLetterObserver(events::add).build().poll(); + FakeSqsClient native2 = new FakeSqsClient(); + sendWithVersion(native2, "not json", "2"); + SqsConsumer.builder(native2, URL).deadLetterObserver(events::add).build().poll(); + + // dead_letter strategy without a DLQ queue degrades to delete. + FakeSqsClient degraded = new FakeSqsClient(); + degraded.seed(URL, validBody(), 1); + SqsConsumer.builder(degraded, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterObserver(events::add).build().poll(); + + // Explicit delete drops poison with no record, even with a DLQ queue configured. + FakeSqsClient dropped = new FakeSqsClient(); + dropped.seed(URL, "{}", 1); + consumerWithDlq(dropped, new ArrayList<>(), new int[1]).unknownUrnStrategy(UnknownUrnStrategy.DELETE) + .deadLetterObserver(events::add).build().poll(); + + assertTrue(events.isEmpty()); + assertEquals(1, degraded.deleted.size()); + assertEquals(1, dropped.deleted.size()); + } + + @Test + void exhaustedRetriesAreLeftToNativeRedriveAndNeverCallTheObserver() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, validBody(), 50); + List events = new ArrayList<>(); + + SqsConsumer.builder(sqs, URL) + .handler("urn:babel:orders:created", (e, m) -> { throw new IllegalStateException("boom"); }) + .deadLetterQueueUrl(DLQ_URL) + .deadLetterObserver(events::add) + .build().poll(); + + assertTrue(events.isEmpty()); + assertTrue(sqs.sent.isEmpty()); + assertTrue(sqs.deleted.isEmpty()); + assertEquals(1, sqs.visibilityChanges.size()); + } + + @Test + void aThrowingObserverNeverChangesTheSettleOutcome() { + for (Throwable failure : new Throwable[] {new IllegalStateException("obs"), new AssertionError("obs")}) { + FakeSqsClient poison = new FakeSqsClient(); + poison.seed(URL, "{}", 1); + List errors = new ArrayList<>(); + consumerWithDlq(poison, errors, new int[1]).deadLetterObserver(e -> sneakyThrow(failure)).build().poll(); + assertEquals(1, poison.sent.size()); + assertEquals(1, poison.deleted.size()); + assertTrue(poison.visibilityChanges.isEmpty()); + assertEquals(1, errors.size()); + + FakeSqsClient gated = new FakeSqsClient(); + sendWithVersion(gated, "not json", "2"); + gated.sent.clear(); + consumerWithDlq(gated, new ArrayList<>(), new int[1]).deadLetterObserver(e -> sneakyThrow(failure)) + .build().poll(); + assertEquals(1, gated.sent.size()); + assertEquals(1, gated.deleted.size()); + + FakeSqsClient unknown = new FakeSqsClient(); + unknown.seed(URL, validBody(), 1); + SqsConsumer.builder(unknown, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).deadLetterObserver(e -> sneakyThrow(failure)).build().poll(); + assertEquals(1, unknown.sent.size()); + assertEquals(1, unknown.deleted.size()); + } + } + + @Test + void withoutAnObserverTheDlqPathIsUnchanged() { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, "{}", 1); + + consumerWithDlq(sqs, new ArrayList<>(), new int[1]).deadLetterObserver(null).build().poll(); + + assertEquals(1, sqs.sent.size()); + assertEquals(1, sqs.deleted.size()); + } + + @SuppressWarnings("unchecked") + private static void sneakyThrow(Throwable t) throws T { + throw (T) t; + } + + @Test + void eventQueueFallbackTreatsOnlyTheAsciiWhitespaceSetAsBlank() { + String json = EnvelopeCodec.encode( + EnvelopeCodec.make("urn:babel:orders:created", Map.of("order_id", 7), "billing", null)); + // meta.queue as JSON text -> the queue the event must report. Only the ASCII whitespace set is blank (ADR-0031): + // Unicode spaces and control characters (Character.isWhitespace) are real, non-blank names. + String[][] cases = { + {" \\t", "orders"}, {"", "orders"}, {"\\u000b\\f\\r", "orders"}, + {"\\u2003", "\u2003"}, {"\\u00a0", "\u00a0"}, {"\\u001c", "\u001c"}, {"\\u0085", "\u0085"}, + }; + for (String[] c : cases) { + FakeSqsClient sqs = new FakeSqsClient(); + sqs.seed(URL, json.replace("\"queue\":\"billing\"", "\"queue\":\"" + c[0] + "\""), 1); + List events = new ArrayList<>(); + SqsConsumer.builder(sqs, URL).unknownUrnStrategy(UnknownUrnStrategy.DEAD_LETTER) + .deadLetterQueueUrl(DLQ_URL).deadLetterObserver(events::add).build().poll(); + assertEquals(1, events.size(), c[0]); + assertEquals(c[1], events.get(0).queue(), "meta.queue=" + c[0]); + } + } } diff --git a/src/test/java/com/babelqueue/sqs/SqsTraceparentTest.java b/src/test/java/com/babelqueue/sqs/SqsTraceparentTest.java index 6afc405..b1db6f2 100644 --- a/src/test/java/com/babelqueue/sqs/SqsTraceparentTest.java +++ b/src/test/java/com/babelqueue/sqs/SqsTraceparentTest.java @@ -155,8 +155,8 @@ void endToEndTraceparentMakesConsumerSpanAChildOfTheProducerSpan() throws Except .poll(); List spans = exporter.getFinishedSpanItems(); - SpanData producer = spanByName(spans, "publish urn:babel:orders:created"); - SpanData consumer = spanByName(spans, "process urn:babel:orders:created"); + SpanData producer = spanByName(spans, "send orders"); + SpanData consumer = spanByName(spans, "process orders"); assertEquals(producer.getSpanContext().getSpanId(), consumer.getParentSpanContext().getSpanId()); assertEquals(producer.getSpanContext().getTraceId(), consumer.getTraceId()); assertTrue(consumer.getParentSpanContext().isRemote());