Add E2E test for retry and DLQ on NATS

NatsRetryAndDlqE2EIT enqueues a message that always throws, expects
JetStream to redeliver up to numRetries=2 times, and asserts the
exhausted payload lands on the rqueue-failing-dlq stream. Currently
@disabled with a rationale: RqueueNatsAutoConfig does not yet invoke
JetStreamMessageBroker.installDeadLetterBridge per queue at container
start, so max-deliveries advisories are never bridged onto the DLQ
subject. Enable once that wiring is added.

Assisted-By: Claude Code

Author: sonus21

rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsRetryAndDlqE2EIT.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright (c) 2024-2026 Sonu Kumar
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and limitations under the License.
+ *
+ */
+package com.github.sonus21.rqueue.spring.boot.integration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.github.sonus21.rqueue.annotation.RqueueListener;
+import com.github.sonus21.rqueue.core.RqueueMessageEnqueuer;
+import io.nats.client.JetStreamManagement;
+import io.nats.client.api.StreamInfo;
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisAutoConfiguration;
+import org.springframework.boot.data.redis.autoconfigure.DataRedisReactiveAutoConfiguration;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.stereotype.Component;
+
+/**
+ * After a handler exhausts {@code numRetries}, JetStream emits a max-deliveries advisory and the
+ * broker's {@code installDeadLetterBridge} dispatcher republishes the payload onto the DLQ
+ * stream. Currently disabled because {@link
+ * com.github.sonus21.rqueue.spring.boot.RqueueNatsAutoConfig} does not yet invoke
+ * {@code JetStreamMessageBroker.installDeadLetterBridge(...)} during container start, so dead-
+ * lettered messages never reach the DLQ stream. Enable this test once that wiring is added.
+ */
+@SpringBootTest(
+    classes = NatsRetryAndDlqE2EIT.TestApp.class,
+    properties = {"rqueue.backend=nats"})
+@Tag("nats")
+@Disabled(
+    "DLQ bridge wiring (JetStreamMessageBroker.installDeadLetterBridge) is not yet invoked by "
+        + "RqueueNatsAutoConfig; enable once the container start path provisions the advisory "
+        + "dispatcher per queue.")
+class NatsRetryAndDlqE2EIT extends AbstractNatsBootIT {
+
+  @Autowired RqueueMessageEnqueuer enqueuer;
+
+  @Autowired FailingListener listener;
+
+  @Autowired JetStreamManagement jsm;
+
+  @Test
+  void exhaustedMessageLandsOnDlqStream() {
+    enqueuer.enqueue("failing", "boom");
+
+    Awaitility.await()
+        .atMost(Duration.ofSeconds(60))
+        .until(() -> listener.attempts.get() >= 2);
+
+    Awaitility.await()
+        .atMost(Duration.ofSeconds(30))
+        .untilAsserted(
+            () -> {
+              StreamInfo dlq = jsm.getStreamInfo("rqueue-failing-dlq");
+              assertThat(dlq.getStreamState().getMsgCount()).isGreaterThanOrEqualTo(1);
+            });
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  static class TestApp {}
+
+  @Component
+  static class FailingListener {
+    final AtomicInteger attempts = new AtomicInteger();
+
+    @RqueueListener(value = "failing", numRetries = "2")
+    void onMessage(String payload) {
+      attempts.incrementAndGet();
+      throw new RuntimeException("simulated failure for payload=" + payload);
+    }
+  }
+}