Do not retry some exceptions

rqueue-core/src/main/java/com/github/sonus21/rqueue/annotation/RqueueListener.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2023 Sonu Kumar
+ * Copyright (c) 2019-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -177,11 +177,18 @@
   String priorityGroup() default "";
 
   /**
-   * The message batch size, Rqueue will try to pull batchSize message in one Redis call. By default
-   * it will try to pull one message but if concurrency is not reached then it will try to pull more
-   * than one messages.
+   * The message batch size, Rqueue will try to pull batchSize message in one Redis call. By
+   * default, it will try to pull one message, but if concurrency is not reached, then it will try to
+   * pull more than one message.
    *
    * @return batch size
    */
   String batchSize() default "-1";
+
+  /**
+   * Exception types that are not retryable, defaults to empty.
+   *
+   * @return exceptions those will not be retried
+   */
+  Class<? extends Throwable>[] doNotRetry() default {};
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023 Sonu Kumar
+ * Copyright (c) 2020-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -16,18 +16,24 @@
 
 package com.github.sonus21.rqueue.listener;
 
+import static com.github.sonus21.rqueue.utils.Constants.ONE_MILLI;
+import static com.github.sonus21.rqueue.utils.Constants.SECONDS_IN_A_MINUTE;
+
 import com.github.sonus21.rqueue.core.RqueueBeanProvider;
 import com.github.sonus21.rqueue.core.middleware.Middleware;
 import com.github.sonus21.rqueue.listener.RqueueMessageListenerContainer.QueueStateMgr;
 import com.github.sonus21.rqueue.utils.QueueThreadPool;
 import com.github.sonus21.rqueue.utils.TimeoutUtils;
+import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import org.slf4j.event.Level;
 import org.springframework.messaging.MessageHeaders;
 
 class DefaultRqueuePoller extends RqueueMessagePoller {
 
+  private Long lastNotAvailableAt;
   private final QueueDetail queueDetail;
   private final QueueThreadPool queueThreadPool;
 
@@ -70,18 +76,31 @@ void deactivate(int index, String queue, DeactivateType deactivateType) {
     }
   }
 
+  private void logNotAvailable() {
+    long maxNotAvailableDelay = 10 * SECONDS_IN_A_MINUTE * ONE_MILLI;
+    if (Objects.isNull(lastNotAvailableAt)) {
+      lastNotAvailableAt = System.currentTimeMillis();
+    } else if (System.currentTimeMillis() - lastNotAvailableAt > maxNotAvailableDelay) {
+      log(Level.ERROR, "deadlock?? frozen?? stuck?? No Threads are available in last {}",
+          null,
+          Duration.ofMillis(maxNotAvailableDelay));
+    }
+    log(Level.DEBUG, "No Threads are available sleeping {}Ms", null, pollingInterval);
+  }
+
   void poll() {
     if (!hasAvailableThreads(queueDetail, queueThreadPool)) {
-      log(Level.WARN, "No Threads are available sleeping {}Ms", null, pollingInterval);
+      logNotAvailable();
       TimeoutUtils.sleepLog(pollingInterval, false);
     } else {
       super.poll(-1, queueDetail.getName(), queueDetail, queueThreadPool);
+      lastNotAvailableAt = null;
     }
   }
 
   @Override
   public void start() {
-    log(Level.DEBUG, "Running Queue {}", null, queueDetail.getName());
+    log(Level.INFO, "poll starting", null);
     while (true) {
       try {
         if (eligibleForPolling(queueDetail.getName())) {
@@ -91,8 +110,8 @@ public void start() {
         } else {
           deactivate(-1, queueDetail.getName(), DeactivateType.NO_MESSAGE);
         }
-      } catch (Exception e) {
-        log(Level.ERROR, "Error in poller", e);
+      } catch (Throwable e) {
+        log(Level.ERROR, "error in polling", e);
         if (shouldExit()) {
           return;
         }

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

@@ -336,7 +336,7 @@ private void saveMessageMetadata(Callable<Void> callable) {
   void updateMessageStatus(MessageStatus messageStatus) {
     setMessageStatus(messageStatus);
     // We need to address these problems with message metadata
-    // 1. Message was deleted while executing, this means local copy is stale
+    // 1. The Message was deleted while executing; this means local copy is stale
     // 2. Parallel update is being made [dashboard operation, periodic job (two periodic jobs can
     // run in parallel due to failure)]
     if (!messageStatus.isTerminalState() || getRqueueMessage().isPeriodic()) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2023 Sonu Kumar
+ * Copyright (c) 2019-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -44,6 +44,7 @@ class MappingInformation implements Comparable<MappingInformation> {
   private final Map<String, Integer> priority;
   private final boolean primary;
   private final int batchSize;
+  private final Set<Class<? extends Throwable>> doNotRetry;
 
   @Override
   public String toString() {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023 Sonu Kumar
+ * Copyright (c) 2020-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -61,7 +61,10 @@ class PostProcessingHandler extends PrefixLogger {
     this.rqueueSystemConfigDao = rqueueSystemConfigDao;
   }
 
-  void handle(JobImpl job, ExecutionStatus status, int failureCount) {
+  void handle(JobImpl job,
+      ExecutionStatus status,
+      int failureCount,
+      Throwable throwable) {
     try {
       switch (status) {
         case QUEUE_INACTIVE:
@@ -79,7 +82,7 @@ void handle(JobImpl job, ExecutionStatus status, int failureCount) {
           handleSuccessFullExecution(job, failureCount);
           break;
         case FAILED:
-          handleFailure(job, failureCount);
+          handleFailure(job, failureCount, throwable);
           break;
         default:
           throw new UnknownSwitchCase(String.valueOf(status));
@@ -153,7 +156,7 @@ private void moveMessageToQueue(
         });
   }
 
-  private void moveMessageToDlq(JobImpl job, int failureCount) {
+  private void moveMessageToDlq(JobImpl job, int failureCount, Throwable throwable) {
     log(
         Level.DEBUG,
         "Message {} Moved to dead letter queue: {}",
@@ -184,7 +187,8 @@ private void moveMessageToDlq(JobImpl job, int failureCount) {
         newMessage.setFailureCount(0);
         newMessage.setSourceQueueName(rqueueMessage.getQueueName());
         newMessage.setSourceQueueFailureCount(failureCount);
-        long backOff = taskExecutionBackoff.nextBackOff(userMessage, newMessage, failureCount);
+        long backOff = taskExecutionBackoff.nextBackOff(userMessage, newMessage, failureCount,
+            throwable);
         backOff =
             (backOff == TaskExecutionBackOff.STOP)
                 ? FixedTaskExecutionBackOff.DEFAULT_INTERVAL
@@ -256,9 +260,9 @@ private void handleSuccessFullExecution(JobImpl job, int failureCount) {
     deleteMessage(job, MessageStatus.SUCCESSFUL, failureCount);
   }
 
-  private void handleRetryExceededMessage(JobImpl job, int failureCount) {
+  private void handleRetryExceededMessage(JobImpl job, int failureCount, Throwable throwable) {
     if (job.getQueueDetail().isDlqSet()) {
-      moveMessageToDlq(job, failureCount);
+      moveMessageToDlq(job, failureCount, throwable);
     } else {
       discardMessage(job, failureCount);
     }
@@ -270,18 +274,19 @@ private int getMaxRetryCount(RqueueMessage rqueueMessage, QueueDetail queueDetai
         : rqueueMessage.getRetryCount();
   }
 
-  private void handleFailure(JobImpl job, int failureCount) {
+  private void handleFailure(JobImpl job, int failureCount, Throwable throwable) {
     int maxRetryCount = getMaxRetryCount(job.getRqueueMessage(), job.getQueueDetail());
     if (failureCount < maxRetryCount) {
       long delay =
-          taskExecutionBackoff.nextBackOff(job.getMessage(), job.getRqueueMessage(), failureCount);
+          taskExecutionBackoff.nextBackOff(job.getMessage(), job.getRqueueMessage(), failureCount,
+              throwable);
       if (delay == TaskExecutionBackOff.STOP) {
-        handleRetryExceededMessage(job, failureCount);
+        handleRetryExceededMessage(job, failureCount, throwable);
       } else {
         parkMessageForRetry(job, null, failureCount, delay);
       }
     } else {
-      handleRetryExceededMessage(job, failureCount);
+      handleRetryExceededMessage(job, failureCount, throwable);
     }
   }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2023 Sonu Kumar
+ * Copyright (c) 2019-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -33,6 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Set;
 import lombok.Builder;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -65,6 +66,7 @@ public class QueueDetail extends SerializableBase {
   private final int batchSize;
   private Map<String, Integer> priority;
   private String priorityGroup;
+  private Set<Class<? extends Throwable>> doNotRetry;
 
   public boolean isDlqSet() {
     return !StringUtils.isEmpty(deadLetterQueueName);
@@ -100,7 +102,7 @@ public QueueConfig toConfig() {
   }
 
   List<QueueDetail> expandQueueDetail(boolean addDefault, int priority) {
-    List<QueueDetail> queueDetails = new ArrayList<>();
+    List<QueueDetail> queueDetails = new ArrayList<>(1 + getPriority().size());
     for (Entry<String, Integer> entry : getPriority().entrySet()) {
       QueueDetail cloneQueueDetail = cloneQueueDetail(entry.getKey(), entry.getValue(), name);
       queueDetails.add(cloneQueueDetail);
@@ -145,6 +147,7 @@ private QueueDetail cloneQueueDetail(
         .priorityGroup(priorityGroup)
         .concurrency(concurrency)
         .priority(Collections.singletonMap(Constants.DEFAULT_PRIORITY_KEY, priority))
+        .doNotRetry(doNotRetry)
         .build();
   }
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2023 Sonu Kumar
+ * Copyright (c) 2020-2024 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * You may not use this file except in compliance with the License.
@@ -35,6 +35,8 @@
 import com.github.sonus21.rqueue.utils.QueueThreadPool;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
+import java.util.Set;
 import org.slf4j.LoggerFactory;
 import org.slf4j.event.Level;
 import org.springframework.messaging.Message;
@@ -111,14 +113,14 @@ private void init() {
   }
 
   private int getMaxRetryCount() {
-    return job.getRqueueMessage().getRetryCount() == null
+    return Objects.isNull(job.getRqueueMessage().getRetryCount())
         ? job.getQueueDetail().getNumRetry()
         : job.getRqueueMessage().getRetryCount();
   }
 
   private void updateCounter(boolean fail) {
     RqueueMetricsCounter counter = beanProvider.getRqueueMetricsCounter();
-    if (counter == null) {
+    if (Objects.isNull(counter)) {
       return;
     }
     if (fail) {
@@ -205,8 +207,7 @@ private void updateToProcessing() {
     this.job.updateMessageStatus(MessageStatus.PROCESSING);
   }
 
-  private void logExecutionTimeWarning(
-      long maxProcessingTime, long startTime, ExecutionStatus status) {
+  private void logExecutionTimeWarning(long maxProcessingTime, long startTime) {
     if (System.currentTimeMillis() > maxProcessingTime) {
       long maxAllowedTime = maxExecutionTime();
       long executionTime = System.currentTimeMillis() - startTime;
@@ -228,11 +229,7 @@ private void begin() {
   }
 
   private void end() {
-    if (status == null) {
-      job.updateExecutionStatus(ExecutionStatus.FAILED, error);
-    } else {
-      job.updateExecutionStatus(status, error);
-    }
+    job.updateExecutionStatus(status, error);
   }
 
   private void callMiddlewares(int currentIndex, List<Middleware> middlewares, Job job)
@@ -252,7 +249,7 @@ private void callMiddlewares(int currentIndex, List<Middleware> middlewares, Job
   }
 
   private void processMessage() throws Exception {
-    if (middlewareList == null) {
+    if (Objects.isNull(middlewareList)) {
       callMiddlewares(0, Collections.emptyList(), job);
     } else {
       callMiddlewares(0, middlewareList, job);
@@ -269,14 +266,30 @@ private void execute() {
       updateCounter(true);
       failureCount += 1;
       error = e;
-    } catch (Exception e) {
+      status = ExecutionStatus.FAILED;
+    } catch (Throwable e) {
       updateCounter(true);
       failureCount += 1;
       error = e;
+      status = ExecutionStatus.FAILED;
       log(Level.ERROR, "Message execution failed, RqueueMessage: {}", e, job.getRqueueMessage());
     }
   }
 
+  private boolean shouldRetry(long maxProcessingTime, int retryCount) {
+    if (retryCount > 0 &&
+        ExecutionStatus.FAILED.equals(status) &&
+        System.currentTimeMillis() < maxProcessingTime) {
+      Set<Class<? extends Throwable>> exceptions = queueDetail.getDoNotRetry();
+      boolean doNoRetry = Objects.nonNull(exceptions) &&
+          !exceptions.isEmpty() &&
+          Objects.nonNull(error) &&
+          exceptions.contains(error.getClass());
+      return !doNoRetry;
+    }
+    return false;
+  }
+
   private void handleMessage() {
     long maxProcessingTime = getMaxProcessingTime();
     long startTime = System.currentTimeMillis();
@@ -285,23 +298,22 @@ private void handleMessage() {
     do {
       log(Level.DEBUG, "Attempt {} message: {}", null, attempt, job.getMessage());
       begin();
-      if (status == null) {
+      if (Objects.isNull(status)) {
         execute();
       }
       retryCount -= 1;
       attempt += 1;
       end();
-    } while (retryCount > 0 && status == null && System.currentTimeMillis() < maxProcessingTime);
-    postProcessingHandler.handle(
-        job, (status == null ? ExecutionStatus.FAILED : status), failureCount);
-    logExecutionTimeWarning(maxProcessingTime, startTime, status);
+    } while (shouldRetry(maxProcessingTime, retryCount));
+    postProcessingHandler.handle(job, status, failureCount, error);
+    logExecutionTimeWarning(maxProcessingTime, startTime);
   }
 
   private long getTtlForScheduledMessageKey(RqueueMessage message) {
     // Assume a message can be executing for at most 2x of their visibility timeout
     // due to failure in some other job same message should not be enqueued
     long expiryInSeconds = 2 * job.getQueueDetail().getVisibilityTimeout() / ONE_MILLI;
-    // A message wil be processed after period, so it must stay in the system till that time
+    // A message wil be processed after a period, so it must stay in the system till that time
     // how many more seconds are left to process this message
     long remainingTime = (message.getProcessAt() - System.currentTimeMillis()) / ONE_MILLI;
     if (remainingTime > 0) {
@@ -311,8 +323,8 @@ private long getTtlForScheduledMessageKey(RqueueMessage message) {
   }
 
   private String getScheduledMessageKey(RqueueMessage message) {
-    // avoid duplicate message enqueue due to retry by checking the message key
-    // avoid cross slot error by using tagged queue name in the key
+    // avoid a duplicate message enqueue due to retry by checking the message key
+    // avoid cross-slot error by using tagged queue name in the key
     // enqueuing duplicate message can lead to duplicate consumption when one job is executing task
     // at the same time this message was enqueued.
     return String.format(