Add global retry limit

Assisted-By: Codex

Author: sonus21

docs/configuration/retry-and-backoff.md

@@ -44,6 +44,17 @@ Note: Messages handled this way are neither retried nor moved to the Dead Letter
 When a message handler fails, the message can be retried immediately, delayed for a 
 future retry, moved to a dead letter queue, or dropped.
 
+### Global Retry Limit
+
+Set `rqueue.retry.max=N` to limit listeners that do not configure `numRetries` and
+would otherwise retry forever. The default is `-1`, which leaves the retry-forever
+default unchanged. Explicit `@RqueueListener(numRetries = "...")` values and the
+dead-letter queue default retry count continue to take precedence.
+
+For the NATS backend, this retry count is translated to JetStream `maxDeliver` as
+`N + 1`, because JetStream counts the initial delivery plus retries. For example,
+`rqueue.retry.max=3` creates consumers with at most four total deliveries.
+
 ### Immediate Retries
 To retry a message immediately within the same polling cycle, set 
 `rqueue.retry.per.poll` to a positive integer (e.g., `2`). This will cause the 
@@ -75,5 +86,3 @@ public class RqueueConfiguration {
 }
 ```
 
-
-

rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueConfig.java

@@ -134,6 +134,9 @@ private static String generateBrokerId() {
   @Value("${rqueue.retry.per.poll:1}")
   private int retryPerPoll;
 
+  @Value("${rqueue.retry.max:-1}")
+  private int maxRetry = -1;
+
   @Value("${rqueue.net.proxy.host:}")
   private String proxyHost;
 

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainer.java

@@ -471,20 +471,20 @@ private AsyncTaskExecutor createTaskExecutor(
 
   private List<QueueDetail> getQueueDetail(String queue, MappingInformation mappingInformation) {
     int numRetry = mappingInformation.getNumRetry();
+    RqueueConfig rqueueConfig = rqueueBeanProvider.getRqueueConfig();
     if (!StringUtils.isEmpty(mappingInformation.getDeadLetterQueueName()) && numRetry == -1) {
       log.warn(
           "Dead letter queue {} is set but retry is not set",
           mappingInformation.getDeadLetterQueueName());
       numRetry = Constants.DEFAULT_RETRY_DEAD_LETTER_QUEUE;
     } else if (numRetry == -1) {
-      numRetry = Integer.MAX_VALUE;
+      numRetry = rqueueConfig.getMaxRetry() >= 0 ? rqueueConfig.getMaxRetry() : Integer.MAX_VALUE;
     }
     String priorityGroup = mappingInformation.getPriorityGroup();
     Map<String, Integer> priority = mappingInformation.getPriority();
     if (StringUtils.isEmpty(priorityGroup) && priority.size() == 1) {
       priorityGroup = Constants.DEFAULT_PRIORITY_GROUP;
     }
-    RqueueConfig rqueueConfig = rqueueBeanProvider.getRqueueConfig();
     QueueDetail queueDetail = QueueDetail.builder()
         .name(queue)
         .queueName(rqueueConfig.getQueueName(queue))

rqueue-core/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -1,6 +1,13 @@
 {
   "groups": [],
   "properties": [
+    {
+      "name": "rqueue.retry.max",
+      "type": "java.lang.Integer",
+      "description": "Global retry limit used when a listener does not configure numRetries and would otherwise retry forever. -1 preserves the retry-forever default. Explicit listener retry counts and DLQ defaults take precedence.",
+      "defaultValue": -1,
+      "sourceType": "com.github.sonus21.rqueue.config.RqueueConfig"
+    },
     {
       "name": "rqueue.nats.auto-create-streams",
       "type": "java.lang.Boolean",

rqueue-core/src/test/java/com/github/sonus21/rqueue/config/RqueueConfigTest.java

@@ -154,4 +154,9 @@ void workerRegistryProperties() {
     assertEquals(
         Duration.ofSeconds(15), rqueueConfigVersion2.getWorkerRegistryQueueHeartbeatInterval());
   }
+
+  @Test
+  void defaultMaxRetryIsDisabled() {
+    assertEquals(-1, rqueueConfigVersion2.getMaxRetry());
+  }
 }

rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainerTest.java

@@ -33,9 +33,11 @@
 import com.github.sonus21.rqueue.common.RqueueLockManager;
 import com.github.sonus21.rqueue.config.RqueueConfig;
 import com.github.sonus21.rqueue.config.RqueueWebConfig;
+import com.github.sonus21.rqueue.core.EndpointRegistry;
 import com.github.sonus21.rqueue.core.RqueueBeanProvider;
 import com.github.sonus21.rqueue.core.RqueueMessage;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
+import com.github.sonus21.rqueue.models.Concurrency;
 import com.github.sonus21.rqueue.dao.RqueueSystemConfigDao;
 import com.github.sonus21.rqueue.models.db.MessageMetadata;
 import com.github.sonus21.rqueue.models.db.QueueConfig;
@@ -67,6 +69,7 @@
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
 
 @CoreUnitTest
 class RqueueMessageListenerContainerTest extends TestBase {
@@ -172,6 +175,62 @@ void setTaskExecutor() {
         ((ThreadPoolTaskExecutor) container.getTaskExecutor()).getThreadNamePrefix());
   }
 
+  @Test
+  void globalMaxRetryCapsDefaultRetryForever() throws Exception {
+    beanProvider.getRqueueConfig().setMaxRetry(5);
+    doReturn(handlerMap(mapping(slowQueue, -1, null))).when(rqueueMessageHandler)
+        .getHandlerMethodMap();
+
+    container.afterPropertiesSet();
+
+    assertEquals(5, EndpointRegistry.get(slowQueue).getNumRetry());
+  }
+
+  @Test
+  void globalMaxRetryDoesNotOverrideExplicitRetryCount() throws Exception {
+    beanProvider.getRqueueConfig().setMaxRetry(2);
+    doReturn(handlerMap(mapping(slowQueue, 10, null))).when(rqueueMessageHandler)
+        .getHandlerMethodMap();
+
+    container.afterPropertiesSet();
+
+    assertEquals(10, EndpointRegistry.get(slowQueue).getNumRetry());
+  }
+
+  @Test
+  void globalMaxRetryDoesNotRaiseLowerExplicitRetryCount() throws Exception {
+    beanProvider.getRqueueConfig().setMaxRetry(5);
+    doReturn(handlerMap(mapping(slowQueue, 2, null))).when(rqueueMessageHandler)
+        .getHandlerMethodMap();
+
+    container.afterPropertiesSet();
+
+    assertEquals(2, EndpointRegistry.get(slowQueue).getNumRetry());
+  }
+
+  @Test
+  void globalMaxRetryCanDisableRetries() throws Exception {
+    beanProvider.getRqueueConfig().setMaxRetry(0);
+    doReturn(handlerMap(mapping(slowQueue, -1, null))).when(rqueueMessageHandler)
+        .getHandlerMethodMap();
+
+    container.afterPropertiesSet();
+
+    assertEquals(0, EndpointRegistry.get(slowQueue).getNumRetry());
+  }
+
+  @Test
+  void globalMaxRetryDoesNotOverrideDeadLetterDefaultRetryCount() throws Exception {
+    beanProvider.getRqueueConfig().setMaxRetry(0);
+    doReturn(handlerMap(mapping(slowQueue, -1, slowQueue + "-dlq"))).when(rqueueMessageHandler)
+        .getHandlerMethodMap();
+
+    container.afterPropertiesSet();
+
+    assertEquals(Constants.DEFAULT_RETRY_DEAD_LETTER_QUEUE,
+        EndpointRegistry.get(slowQueue).getNumRetry());
+  }
+
   @Test
   void phaseSetting() {
     assertEquals(Integer.MAX_VALUE, container.getPhase());
@@ -534,6 +593,28 @@ private class TestListenerContainer extends RqueueMessageListenerContainer {
     }
   }
 
+  private MultiValueMap<MappingInformation, RqueueMessageHandler.HandlerMethodWithPrimary>
+      handlerMap(MappingInformation mappingInformation) {
+    LinkedMultiValueMap<MappingInformation, RqueueMessageHandler.HandlerMethodWithPrimary> map =
+        new LinkedMultiValueMap<>();
+    map.add(mappingInformation, new RqueueMessageHandler.HandlerMethodWithPrimary(null, false));
+    return map;
+  }
+
+  private MappingInformation mapping(String queue, int numRetry, String deadLetterQueueName) {
+    return MappingInformation.builder()
+        .queueNames(Collections.singleton(queue))
+        .numRetry(numRetry)
+        .deadLetterQueueName(deadLetterQueueName)
+        .deadLetterConsumerEnabled(false)
+        .visibilityTimeout(VISIBILITY_TIMEOUT)
+        .active(false)
+        .concurrency(new Concurrency(0, 0))
+        .priority(Collections.singletonMap(Constants.DEFAULT_PRIORITY_KEY, 1))
+        .batchSize(1)
+        .build();
+  }
+
   @Getter
   private class StubMessageSchedulerListenerContainer extends RqueueMessageListenerContainer {
 

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/JetStreamMessageBroker.java

@@ -510,15 +510,16 @@ public static Duration resolveAckWait(QueueDetail q, RqueueNatsConfig config) {
   /**
    * Resolve the JetStream {@code maxDeliver} from per-queue {@link QueueDetail#getNumRetry()}
    * (counted as initial delivery + N retries = numRetry + 1). The {@link Integer#MAX_VALUE}
-   * "retry forever" sentinel maps to JetStream's unlimited value ({@code -1}); non-positive
-   * numRetry falls back to {@code RqueueNatsConfig.ConsumerDefaults.getMaxDeliver()}.
+   * "retry forever" sentinel maps to JetStream's unlimited value ({@code -1}); zero means
+   * one total delivery and no retries. Negative numRetry falls back to
+   * {@code RqueueNatsConfig.ConsumerDefaults.getMaxDeliver()}.
    */
   public static long resolveMaxDeliver(QueueDetail q, RqueueNatsConfig config) {
     int numRetry = q.getNumRetry();
     if (numRetry == Integer.MAX_VALUE) {
       return -1L;
     }
-    if (numRetry > 0) {
+    if (numRetry >= 0) {
       return numRetry + 1L;
     }
     return config.getConsumerDefaults().getMaxDeliver();

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/js/JetStreamMessageBrokerResolveTest.java

@@ -71,10 +71,9 @@ void resolveMaxDeliver_isNumRetryPlusOne() {
   }
 
   @Test
-  void resolveMaxDeliver_fallsBackToConfigDefaultWhenZero() {
+  void resolveMaxDeliver_zeroRetryMeansOneDelivery() {
     QueueDetail q = queue(30_000L, 0);
-    // RqueueNatsConfig.defaults().consumerDefaults.maxDeliver = 3
-    assertEquals(3L, JetStreamMessageBroker.resolveMaxDeliver(q, RqueueNatsConfig.defaults()));
+    assertEquals(1L, JetStreamMessageBroker.resolveMaxDeliver(q, RqueueNatsConfig.defaults()));
   }
 
   @Test

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

@@ -0,0 +1,101 @@
+/*
+ * Copyright (c) 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 java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeAll;
+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.context.annotation.Import;
+import org.springframework.stereotype.Component;
+
+@SpringBootTest(
+    classes = NatsGlobalRetryLimitE2EIT.TestApp.class,
+    properties = {
+      "rqueue.backend=nats",
+      "rqueue.retry.max=2",
+      "rqueue.retry.per.poll=1000",
+      "rqueue.nats.naming.stream-prefix=" + NatsGlobalRetryLimitE2EIT.STREAM_PREFIX,
+      "rqueue.nats.naming.subject-prefix=" + NatsGlobalRetryLimitE2EIT.SUBJECT_PREFIX
+    })
+@Tag("nats")
+class NatsGlobalRetryLimitE2EIT extends AbstractNatsBootIT {
+
+  static final String QUEUE = "global-retry-nats";
+  static final String STREAM_PREFIX = "rqueue-js-globalRetryE2E-";
+  static final String SUBJECT_PREFIX = "rqueue.js.globalRetryE2E.";
+
+  @BeforeAll
+  static void wipeOwnedStreams() {
+    deleteStreamsWithPrefix(STREAM_PREFIX);
+  }
+
+  @Autowired
+  RqueueMessageEnqueuer enqueuer;
+
+  @Autowired
+  JetStreamManagement jsm;
+
+  @Autowired
+  FailingListener listener;
+
+  @Test
+  void globalRetryLimitCapsImplicitRetryForever() throws Exception {
+    enqueuer.enqueue(QUEUE, "payload");
+
+    assertThat(listener.twoAttempts.await(20, TimeUnit.SECONDS)).isTrue();
+    Awaitility.await().during(Duration.ofSeconds(2)).atMost(Duration.ofSeconds(3)).untilAsserted(
+        () -> assertThat(listener.attempts).hasValue(2));
+
+    assertThat(jsm
+            .getConsumerInfo(STREAM_PREFIX + QUEUE, QUEUE + "-consumer")
+            .getConsumerConfiguration()
+            .getMaxDeliver())
+        .isEqualTo(3L);
+  }
+
+  @SpringBootApplication(
+      exclude = {DataRedisAutoConfiguration.class, DataRedisReactiveAutoConfiguration.class})
+  @Import(FailingListener.class)
+  static class TestApp {}
+
+  @Component
+  static class FailingListener {
+    final AtomicInteger attempts = new AtomicInteger();
+    final CountDownLatch twoAttempts = new CountDownLatch(2);
+
+    @RqueueListener(value = QUEUE)
+    void onMessage(String ignored) {
+      attempts.incrementAndGet();
+      twoAttempts.countDown();
+      throw new IllegalStateException("force retry");
+    }
+  }
+}