Add global retry limit (#298)

* Add global retry limit

Assisted-By: Codex

* Apply Palantir Java Format

* Bump version to RC9

Assisted-By: Codex

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: sonus21

build.gradle

@@ -84,7 +84,7 @@ ext {
 
 subprojects {
     group = "com.github.sonus21"
-    version = "4.0.0-RC8"
+    version = "4.0.0-RC9"
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

docs/CHANGELOG.md

@@ -18,6 +18,18 @@ foundational Spring Boot 4 and Jackson 3 migration notes; RC3 for the Java 17
 baseline change; RC4–RC6 below for the NATS backend, broker SPI, dashboard
 work, and middleware additions that build on top.
 
+## Release [4.0.0.RC9] 2026-05-13
+
+{: .highlight}
+Release candidate.
+
+### Features
+* **Global retry limit** — added `rqueue.retry.max` to cap the implicit
+  retry-forever default for listeners that do not configure `numRetries`.
+  Explicit per-listener retry counts and the existing DLQ retry default continue
+  to take precedence. On NATS, the effective retry count maps to JetStream
+  `maxDeliver` as `retries + 1`.
+
 ## Release [4.0.0.RC8] 2026-05-09
 
 {: .highlight}

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,10 +33,12 @@
 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.dao.RqueueSystemConfigDao;
+import com.github.sonus21.rqueue.models.Concurrency;
 import com.github.sonus21.rqueue.models.db.MessageMetadata;
 import com.github.sonus21.rqueue.models.db.QueueConfig;
 import com.github.sonus21.rqueue.models.enums.MessageStatus;
@@ -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,68 @@ 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 +599,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