Add global retry limit (#298)

sonus21 · github-actions[bot] · web-flow · commit 2769b64e903a · 2026-05-16T00:02:31.000+05:30
* Add global retry limit

Assisted-By: Codex

* Apply Palantir Java Format

* Bump version to RC9

Assisted-By: Codex

---------

Co-authored-by: github-actions[bot] &lt;41898282+github-actions[bot]@users.noreply.github.com&gt;
diff --git a/build.gradle b/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
diff --git a/docs/CHANGELOG.md b/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}
diff --git a/docs/configuration/retry-and-backoff.md b/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 {
 }
 ```
  
-
-
diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/config/RqueueConfig.java b/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;
 
diff --git a/rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainer.java b/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))
diff --git a/rqueue-core/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/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",
diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/config/RqueueConfigTest.java b/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());
+  }
 }
diff --git a/rqueue-core/src/test/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainerTest.java b/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 {
 
diff --git a/rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/JetStreamMessageBroker.java b/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();
diff --git a/rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/js/JetStreamMessageBrokerResolveTest.java b/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
diff --git a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsGlobalRetryLimitE2EIT.java b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/integration/NatsGlobalRetryLimitE2EIT.java
@@ -0,0 +1,102 @@
+/*
+ * 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");
+    }
+  }
+}
diff --git a/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RedisGlobalRetryLimitE2EIT.java b/rqueue-spring-boot-starter/src/test/java/com/github/sonus21/rqueue/spring/boot/tests/integration/RedisGlobalRetryLimitE2EIT.java

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,13 @@`
`1`	`1`	`{`
`2`	`2`	`"groups": [],`
`3`	`3`	`"properties": [`
	`4`	`+ {`
	`5`	`+ "name": "rqueue.retry.max",`
	`6`	`+ "type": "java.lang.Integer",`
	`7`	`+ "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.",`
	`8`	`+ "defaultValue": -1,`
	`9`	`+ "sourceType": "com.github.sonus21.rqueue.config.RqueueConfig"`
	`10`	`+ },`
`4`	`11`	`{`
`5`	`12`	`"name": "rqueue.nats.auto-create-streams",`
`6`	`13`	`"type": "java.lang.Boolean",`
Original file line number	Diff line number	Diff line change
`@@ -154,4 +154,9 @@ void workerRegistryProperties() {`
`154`	`154`	`assertEquals(`
`155`	`155`	`Duration.ofSeconds(15), rqueueConfigVersion2.getWorkerRegistryQueueHeartbeatInterval());`
`156`	`156`	`}`
	`157`	`+`
	`158`	`+ @Test`
	`159`	`+ void defaultMaxRetryIsDisabled() {`
	`160`	`+ assertEquals(-1, rqueueConfigVersion2.getMaxRetry());`
	`161`	`+ }`
`157`	`162`	`}`
Original file line number	Diff line number	Diff line change
`@@ -71,10 +71,9 @@ void resolveMaxDeliver_isNumRetryPlusOne() {`
`71`	`71`	`}`
`72`	`72`
`73`	`73`	`@Test`
`74`		`- void resolveMaxDeliver_fallsBackToConfigDefaultWhenZero() {`
	`74`	`+ void resolveMaxDeliver_zeroRetryMeansOneDelivery() {`
`75`	`75`	`QueueDetail q = queue(30_000L, 0);`
`76`		`- // RqueueNatsConfig.defaults().consumerDefaults.maxDeliver = 3`
`77`		`- assertEquals(3L, JetStreamMessageBroker.resolveMaxDeliver(q, RqueueNatsConfig.defaults()));`
	`76`	`+ assertEquals(1L, JetStreamMessageBroker.resolveMaxDeliver(q, RqueueNatsConfig.defaults()));`
`78`	`77`	`}`
`79`	`78`
`80`	`79`	`@Test`