Fixing bug #170 #169 (#171)

Fixing bug #170 #169

Author: sonus21

CHANGELOG.md

@@ -1,4 +1,11 @@
 # [Rqueue] New and Notable Changes
+### [2.11] - 07-Nov-2022
+
+### Fixes
+
+* Message mover unreliability, scheduled message were not getting consumed once redis connection error occurs
+* Upgraded Jquery version
+
 
 ### [2.10.2] - 16-Jul-2022
 
@@ -295,4 +302,6 @@ Fixes:
 
 [2.10.2]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.10.2-RELEASE
 
+[2.11]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.11-RELEASE
+
 [122]: https://github.com/sonus21/rqueue/issues/122

README.md

@@ -71,14 +71,14 @@ Release Version: [Maven central](https://search.maven.org/search?q=g:com.github.
 * Add dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.10.2-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.11-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
         <groupId>com.github.sonus21</groupId>
         <artifactId>rqueue-spring-boot-starter</artifactId>
-        <version>2.10.2-RELEASE</version>
+        <version>2.11-RELEASE</version>
     </dependency>
     ```
 
@@ -91,14 +91,14 @@ Release Version: [Maven central](https://search.maven.org/search?q=g:com.github.
 * Add Dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring:2.10.2-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring:2.11-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
        <groupId>com.github.sonus21</groupId>
        <artifactId>rqueue-spring</artifactId>
-       <version>2.10.2-RELEASE</version>
+       <version>2.11-RELEASE</version>
      </dependency>
     ```
 

build.gradle

@@ -70,7 +70,7 @@ ext {
 
 subprojects {
     group = 'com.github.sonus21'
-    version = '2.10.2-RELEASE'
+    version = '2.11-RELEASE'
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

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

@@ -65,6 +65,11 @@ public class RqueueSchedulerConfig {
   private int processingMessageThreadPoolSize;
 
   // How frequently messages should be moved from scheduled queues to source queue
-  @Value("${rqueue.scheduler.scheduled.message.time.interval:5000}")
+  @Value("${rqueue.scheduler.scheduled.message.time.interval:2000}")
   private long scheduledMessageTimeIntervalInMilli;
+
+
+  // Maximum delay for message mover task due to failure
+  @Value("${rqueue.scheduler.max.message.mover.delay:60000}")
+  private long maxMessageMoverDelay;
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/MessageScheduler.java

@@ -73,6 +73,7 @@ public abstract class MessageScheduler
   private RedisTemplate<String, Long> redisTemplate;
 
   private Map<String, QueueScheduler> queueSchedulers;
+  private Map<String, Integer> errorCount;
 
   protected abstract Logger getLogger();
 
@@ -212,6 +213,7 @@ protected void initialize() {
     channelNameToQueueName = new ConcurrentHashMap<>(queueNames.size());
     queueNameToLastMessageScheduleTime = new ConcurrentHashMap<>(queueNames.size());
     queueSchedulers = new ConcurrentHashMap<>(queueNames.size());
+    errorCount = new ConcurrentHashMap<>(queueNames.size());
     createScheduler(queueNames.size());
     if (isRedisEnabled()) {
       messageSchedulerListener = new MessageSchedulerListener();
@@ -362,26 +364,50 @@ private class MessageMoverTask implements Runnable {
     private final String zsetName;
     private final boolean processingQueue;
 
+    private long getNextScheduleTimeInternal(Long value, Exception e) {
+      int errCount = 0;
+      long nextTime;
+      if (null != e) {
+        errCount = errorCount.getOrDefault(queueName, 0) + 1;
+        if (errCount % 3 == 0) {
+          getLogger().error("Message mover task is failing continuously queue: {}", name, e);
+        }
+        long delay = (long) (100 * Math.pow(1.5, errCount));
+        delay = Math.min(delay, rqueueSchedulerConfig.getMaxMessageMoverDelay());
+        nextTime = System.currentTimeMillis() + delay;
+      }else{
+        nextTime = getNextScheduleTime(queueName, value);
+      }
+      errorCount.put(queueName, errCount);
+      return nextTime;
+    }
+
     @Override
     public void run() {
       getLogger().debug("Running {}", this);
+      Long value = null;
+      Exception e = null;
       try {
         if (isQueueActive(name)) {
           long currentTime = System.currentTimeMillis();
-          Long value =
+          value =
               defaultScriptExecutor.execute(
                   redisScript,
                   Arrays.asList(queueName, zsetName),
                   currentTime,
                   MAX_MESSAGES,
                   processingQueue ? 1 : 0);
-          long nextExecutionTime = getNextScheduleTime(name, value);
+        }
+      } catch (RedisSystemException ex) {
+        e = ex;
+      } catch (Exception ex) {
+        e = ex;
+        getLogger().warn("Task execution failed for the queue: {}", getName(), e);
+      } finally {
+        if (isQueueActive(name)) {
+          long nextExecutionTime = getNextScheduleTimeInternal(value, e);
           schedule(name, nextExecutionTime, true);
         }
-      } catch (RedisSystemException e) {
-        // no op
-      } catch (Exception e) {
-        getLogger().warn("Task execution failed for the queue: {}", name, e);
       }
     }
 

rqueue-core/src/main/resources/public/rqueue/vendor/jquery/jquery.min.js

@@ -52,8 +52,13 @@
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
+import org.springframework.data.redis.ClusterRedirectException;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.RedisSystemException;
+import org.springframework.data.redis.TooManyClusterRedirectionsException;
 import org.springframework.data.redis.connection.DefaultMessage;
 import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.connection.RedisInvalidSubscriptionException;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.listener.ChannelTopic;
@@ -62,15 +67,20 @@
 @SuppressWarnings("unchecked")
 class ScheduledQueueMessageSchedulerTest extends TestBase {
 
-  @InjectMocks private final TestMessageScheduler messageScheduler = new TestMessageScheduler();
+  @InjectMocks
+  private final TestMessageScheduler messageScheduler = new TestMessageScheduler();
   private final String slowQueue = "slow-queue";
   private final String fastQueue = "fast-queue";
   private final QueueDetail slowQueueDetail = TestUtils.createQueueDetail(slowQueue);
   private final QueueDetail fastQueueDetail = TestUtils.createQueueDetail(fastQueue);
-  @Mock private RqueueSchedulerConfig rqueueSchedulerConfig;
-  @Mock private RqueueConfig rqueueConfig;
-  @Mock private RedisTemplate<String, Long> redisTemplate;
-  @Mock private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory;
+  @Mock
+  private RqueueSchedulerConfig rqueueSchedulerConfig;
+  @Mock
+  private RqueueConfig rqueueConfig;
+  @Mock
+  private RedisTemplate<String, Long> redisTemplate;
+  @Mock
+  private RqueueRedisListenerContainerFactory rqueueRedisListenerContainerFactory;
 
   @BeforeEach
   public void init() {
@@ -230,10 +240,10 @@ void startSubmitsTaskAndThatGetsExecuted() throws Exception {
     doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli();
     AtomicInteger counter = new AtomicInteger(0);
     doAnswer(
-            invocation -> {
-              counter.incrementAndGet();
-              return null;
-            })
+        invocation -> {
+          counter.incrementAndGet();
+          return null;
+        })
         .when(redisTemplate)
         .execute(any(RedisCallback.class));
     messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true));
@@ -251,10 +261,10 @@ void onCompletionOfExistingTaskNewTaskShouldBeSubmitted() throws Exception {
       doReturn(1000L).when(rqueueSchedulerConfig).getScheduledMessageTimeIntervalInMilli();
       AtomicInteger counter = new AtomicInteger(0);
       doAnswer(
-              invocation -> {
-                counter.incrementAndGet();
-                return null;
-              })
+          invocation -> {
+            counter.incrementAndGet();
+            return null;
+          })
           .when(redisTemplate)
           .execute(any(RedisCallback.class));
       TestTaskScheduler scheduler = new TestTaskScheduler();
@@ -269,6 +279,68 @@ void onCompletionOfExistingTaskNewTaskShouldBeSubmitted() throws Exception {
     }
   }
 
+  @Test
+  void taskShouldBeScheduledOnFailure() throws Exception {
+    try (MockedStatic<ThreadUtils> threadUtils = Mockito.mockStatic(ThreadUtils.class)) {
+      doReturn(1).when(rqueueSchedulerConfig).getScheduledMessageThreadPoolSize();
+      doReturn(true).when(rqueueSchedulerConfig).isAutoStart();
+      doReturn(true).when(rqueueSchedulerConfig).isEnabled();
+      doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled();
+      doReturn(10000L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay();
+      AtomicInteger counter = new AtomicInteger(0);
+      doAnswer(
+          invocation -> {
+            counter.incrementAndGet();
+            throw new RedisSystemException("Something is not correct", new NullPointerException("oops!"));
+          })
+          .when(redisTemplate)
+          .execute(any(RedisCallback.class));
+      TestTaskScheduler scheduler = new TestTaskScheduler();
+      threadUtils
+          .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60))
+          .thenReturn(scheduler);
+      messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true));
+      waitFor(() -> counter.get() >= 1, "scripts are getting executed");
+      sleep(10);
+      messageScheduler.destroy();
+      assertTrue(scheduler.submittedTasks() >= 2);
+    }
+  }
+
+  @Test
+  void continuousTaskFailTask() throws Exception{
+    try (MockedStatic<ThreadUtils> threadUtils = Mockito.mockStatic(ThreadUtils.class)) {
+      doReturn(1).when(rqueueSchedulerConfig).getScheduledMessageThreadPoolSize();
+      doReturn(true).when(rqueueSchedulerConfig).isAutoStart();
+      doReturn(true).when(rqueueSchedulerConfig).isEnabled();
+      doReturn(true).when(rqueueSchedulerConfig).isRedisEnabled();
+      doReturn(100L).when(rqueueSchedulerConfig).getMaxMessageMoverDelay();
+      AtomicInteger counter = new AtomicInteger(0);
+      doAnswer(
+          invocation -> {
+            int count = counter.incrementAndGet();
+            if(count % 3 == 0){
+              throw new RedisSystemException("Something is not correct", new NullPointerException("oops!"));
+            }
+            if(count % 3 == 1){
+              throw  new RedisConnectionFailureException("Unknown host");
+            }
+            throw new ClusterRedirectException(3, "localhost", 9004, new TooManyClusterRedirectionsException("too many redirects") );
+          })
+          .when(redisTemplate)
+          .execute(any(RedisCallback.class));
+      TestTaskScheduler scheduler = new TestTaskScheduler();
+      threadUtils
+          .when(() -> ThreadUtils.createTaskScheduler(1, "scheduledQueueMsgScheduler-", 60))
+          .thenReturn(scheduler);
+      messageScheduler.onApplicationEvent(new RqueueBootstrapEvent("Test", true));
+      waitFor(() -> counter.get() >= 10, "scripts are getting executed");
+      sleep(10);
+      messageScheduler.destroy();
+      assertTrue(scheduler.submittedTasks() >= 11);
+    }
+  }
+
   @Test
   void onMessageListenerTest() throws Exception {
     doReturn(true).when(rqueueSchedulerConfig).isEnabled();