Fixing #193.

README.md

@@ -80,14 +80,14 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN
 * Add dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.0-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.1-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
         <groupId>com.github.sonus21</groupId>
         <artifactId>rqueue-spring-boot-starter</artifactId>
-        <version>2.13.0-RELEASE</version>
+        <version>2.13.1-RELEASE</version>
     </dependency>
     ```
 
@@ -108,14 +108,14 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN
 * Add Dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring:2.13.0-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring:2.13.1-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
        <groupId>com.github.sonus21</groupId>
        <artifactId>rqueue-spring</artifactId>
-       <version>2.13.0-RELEASE</version>
+       <version>2.13.1-RELEASE</version>
      </dependency>
     ```
 * Add annotation `EnableRqueue` on application config class
@@ -127,7 +127,6 @@ from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AN
 
 @EnableRqueue
 public class Application {
-
   @Bean
   public RedisConnectionFactory redisConnectionFactory() {
     // return a redis connection factory
@@ -321,13 +320,6 @@ to familiarise yourself with the project.
 
 **Please format your code with Google Java formatter.**
 
-```groovy
-// springBootVersion = '2.0.6.RELEASE'
-// springVersion = '5.0.10.RELEASE'
-// springDataVersion = '2.0.6.RELEASE'
-// microMeterVersion = '1.1.0'
-```
-
 ## Links
 
 * Documentation: [https://sonus21.github.io/rqueue](https://sonus21.github.io/rqueue)

build.gradle

@@ -25,11 +25,6 @@ allprojects {
 }
 
 ext {
-    springBootVersion = System.getenv("SPRING_BOOT_VERSION")
-    springVersion = System.getenv("SPRING_VERSION")
-    springDataVersion = System.getenv("SPRING_DATA_VERSION")
-    microMeterVersion = System.getenv("MICROMETER_VERSION")
-
     springBootVersion = "3.0.1"
     springVersion = "6.0.3"
     springDataVersion = "3.0.0"
@@ -77,7 +72,7 @@ ext {
 
 subprojects {
     group = "com.github.sonus21"
-    version = "3.0.1-RELEASE"
+    version = "3.1.0-RELEASE"
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging

docs/CHANGELOG.md

@@ -8,22 +8,37 @@ layout: default
 
 All notable user-facing changes to this project are documented in this file.
 
-## Release [3.0.1] 17-Jan-2022
+## Release [3.1.0] 24-June-2023
+{: .highlight } 
+Migrate to this version to reduce redis resource utilization
+
+This will fix an important bug happening due to task multiplications. This is causing more Redis
+resource usage Please check #[193]
+
+## Release [2.13.1] 24-June-2023
+{: .highlight }
+Migrate to this version to reduce redis resource utilization
+
+This will fix an important bug happening due to task multiplications. This is causing more Redis
+resource usage Please check #[193]
 
-We're so excited to release Rqueue `3.0.1`. This release supports the Java 17, Spring Boot 3.x and Spring Framework 6.x
 
+## Release [3.0.1] 17-Jan-2022
+
+We're so excited to release Rqueue `3.0.1`. This release supports the Java 17, Spring Boot 3.x and
+Spring Framework 6.x
 
 ### [2.13.0] - 25-Dec-2022
 
 ### Fixes
-{: .highlight} 
+
+{: .highlight}
 Migrate to this version as soon as possible to avoid duplicate message consumption post deletion.
 
 * Important fix for parallel message deletion or delete the message from message listener
 * No threads are available, improvement on message poller
 * Use System Zone ID for UI bottom screen
 
-
 ### [2.12.0] - 14-Dec-2022
 
 ### Fixes
@@ -33,15 +48,14 @@ Migrate to this version as soon as possible to avoid duplicate message consumpti
 
 ### [2.11.1] - 18-Nov-2022
 
-{: .highlight} 
-Migrate to this version as soon as possible to avoid message build up. Messages in scheduled queue 
-can grow if poller is failing. Workaround is to restart the application. 
+{: .highlight}
+Migrate to this version as soon as possible to avoid message build up. Messages in scheduled queue
+can grow if poller is failing. Workaround is to restart the application.
 
-* Message mover unreliability, scheduled message were not getting consumed once redis connection error occurs
+* Message mover unreliability, scheduled message were not getting consumed once redis connection
+  error occurs
 * Upgraded Jquery version
 
-
-
 ### [2.10.2] - 16-Jul-2022
 
 ### Fixes
@@ -60,10 +74,10 @@ can grow if poller is failing. Workaround is to restart the application.
 
 ### [2.10.0] - 10-Oct-2021
 
-{: .warning } 
+{: .warning }
 Breaking change, if you're controlling any internal settings of Rqueue using application environment
-or configuration variable than application can break. We've renamed some config keys, [see](./migration#290-to-210)
-
+or configuration variable than application can break. We've renamed some config
+keys, [see](./migration#290-to-210)
 
 ### Fixes
 
@@ -235,7 +249,6 @@ Breaking change, for migration [see](./migration#1x-to-2x)
   REDIS with prefix `rqueue-` then it will consider version 2.
 - Renamed annotation field `maxJobExecutionTime` to `visibilityTimeout`
 
-
 ### Added
 
 - Web interface to visualize queue
@@ -294,7 +307,6 @@ Breaking change, for migration [see](./migration#1x-to-2x)
 
 * The basic version of Asynchronous task execution using Redis for Spring and Spring Boot
 
-
 [1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/1.0-RELEASE
 
 [1.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue/1.1-RELEASE
@@ -353,6 +365,11 @@ Breaking change, for migration [see](./migration#1x-to-2x)
 
 [2.13.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.13.0-RELEASE
 
+[2.13.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.13.1-RELEASE
+
 [3.0.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/3.0.0-RELEASE
 
+[3.1.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/3.1.0-RELEASE
+
 [122]: https://github.com/sonus21/rqueue/issues/122
+[193]: https://github.com/sonus21/rqueue/issues/193

docs/index.md

@@ -90,38 +90,38 @@ the latest version in your dependency manager.
 
 #### Spring Boot 2.x Setup
 
-Add Rqueue Spring Boot Starter 2.9.0 and refresh your project. Once you've added the dependency,
+Add Rqueue Spring Boot Starter 2.13.1 and refresh your project. Once you've added the dependency,
 you can start sending and consuming messages.
 
 * Gradle
   ```groovy
-      implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.9.0-RELEASE'
+      implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.1-RELEASE'
   ```
 * Maven
   ```xml
    <dependency>
       <groupId>com.github.sonus21</groupId>
       <artifactId>rqueue-spring-boot-starter</artifactId>
-      <version>2.9.0-RELEASE</version>
+      <version>2.13.1-RELEASE</version>
   </dependency>
   ```
 
 #### Spring Boot 3.x Setup
 
-Add Rqueue Spring Boot Starter 3.0.1 and refresh your project. Once you've added the dependency,
+Add Rqueue Spring Boot Starter 3.1.0 and refresh your project. Once you've added the dependency,
 you can start sending and consuming messages.
 
 * Add dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring-boot-starter:3.0.1-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring-boot-starter:3.1.0-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
         <groupId>com.github.sonus21</groupId>
         <artifactId>rqueue-spring-boot-starter</artifactId>
-        <version>3.0.1-RELEASE</version>
+        <version>3.1.0-RELEASE</version>
     </dependency>
     ```
 
@@ -140,29 +140,29 @@ in your dependency manager.
 
 * Gradle
   ```groovy
-      implementation 'com.github.sonus21:rqueue-spring:2.9.0-RELEASE'
+      implementation 'com.github.sonus21:rqueue-spring:2.13.1-RELEASE'
   ```
 * Maven
   ```xml
    <dependency>
      <groupId>com.github.sonus21</groupId>
      <artifactId>rqueue-spring</artifactId>
-     <version>2.9.0-RELEASE</version>
+     <version>2.13.1-RELEASE</version>
    </dependency>
   ```
 
 #### Spring Framework 6.x
 
 * Gradle
   ```groovy
-      implementation 'com.github.sonus21:rqueue-spring:3.0.1-RELEASE'
+      implementation 'com.github.sonus21:rqueue-spring:3.1.0-RELEASE'
   ```
 * Maven
   ```xml
    <dependency>
      <groupId>com.github.sonus21</groupId>
      <artifactId>rqueue-spring</artifactId>
-     <version>3.0.1-RELEASE</version>
+     <version>3.1.0-RELEASE</version>
    </dependency>
   ```
 
@@ -314,6 +314,8 @@ Rqueue is stable and production ready, it's processing millions of on messages d
 [![Line](static/users/line.png){: width="70" height="60" alt="Line Chat" }](https://line.me){:target="_blank" style="margin:10px"}
 [![Aviva](static/users/aviva.jpeg){: width="70" height="60" alt="Aviva" }](https://www.aviva.com/){:target="_blank" style="margin:10px"}
 [![Diamler Truck](static/users/mercedes.png){: width="80" height="60" alt="Daimler Truck (Mercedes)" }](https://www.daimlertruck.com/en){:target="_blank" style="margin:10px"}
+[![Vonage](static/users/vonage.png){: width="250" height="60" alt="Vonage" }](http://vonage.com){:target="_blank" style="margin:10px"}
+[![T-Mobile](static/users/t-mobile.svg){: width="50" height="50" alt="T Mobile" }](https://www.t-mobile.com){:target="_blank" style="margin:10px"}
 [![Poker Stars](static/users/pokerstars.png){: width="210" height="60" alt="PokerStars" }](https://www.pokerstarssports.eu){:target="_blank" style="margin:10px"}
 [![Tune You](static/users/tuneyou.png){: width="140" height="60" alt="TuneYou" }](https://tuneyou.com){:target="_blank" style="margin:10px"}
 [![Bit bot](static/users/bitbot.png){: width="80" height="60" alt="BitBot" }](https://bitbot.plus){:target="_blank" style="margin:10px"}

docs/message-handling/producer-consumer.md

@@ -163,17 +163,29 @@ supported configurations.
   setting.
 * `rqueue.scheduler.auto.start=true` Rqueue scheduler also have thread pools, that handles the
   message. If you would like to use only event based message movement than set auto start as false.
-* `rqueue.scheduler.scheduled.message.thread.pool.size=5` There could be many delayed queues, in that
-  case Rqueue has to move more messages from ZSET to LIST. In such cases, you can increase thread
-  pool size, the number of threads used for message movement is minimum of queue count and pool
-  size.
+* `rqueue.scheduler.scheduled.message.thread.pool.size=5` There could be many delayed queues, in
+  that case Rqueue has to move more messages from ZSET to LIST. In such cases, you can increase
+  thread pool size, the number of threads used for message movement is minimum of queue count and
+  pool size.
 * `rqueue.scheduler.processing.message.thread.pool.size=1` there could be some dead message in
   processing queue as well, if you're seeing large number of dead messages in processing queue, then
   you should increase the thread pool size. Processing queue is used for at least once message
   delivery guarantee.
 * `rqueue.scheduler.scheduled.message.time.interval=5000` At what interval message should be moved
   from scheduled queue to normal queue. The default value is 5 seconds, that means, you can observe
   minimum delay of 5 seconds in delayed message consumption.
+* `rqueue.scheduler.max.message.count=100` Rqueue continuously move scheduled messages from
+  processing/scheduled queue to normal queue so that we can process them asap. There are many
+  instances when large number of messages are scheduled to be run in next 5 minutes. In such cases
+  Rqueue can pull message from scheduled queue to normal queue at higher rate. By default, it copies
+  100 messages from scheduled/processing queue to normal queue.
+* `rqueue.scheduler.max.message.mover.delay=60000` Rqueue continuously move scheduled messages from
+  processing/scheduled queue to normal queue so that we can process them asap. Due to failure, it
+  can load the Redis system, in such cases it uses exponential backoff to limit the damage. This
+  time indicates maximum time for which it should wait before making Redis calls.
+* `rqueue.scheduler.min.message.mover.delay=200` Rqueue continuously move scheduled messages from
+  processing/scheduled queue to normal queue so that we can process them asap. It periodically
+  fetches the messages, the minium delay in such cases can be configured using this variable.
 
 ### Dead Letter Queue Consumer/Listener
 

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

@@ -16,6 +16,7 @@
 
 package com.github.sonus21.rqueue.config;
 
+import com.github.sonus21.rqueue.utils.Constants;
 import lombok.Getter;
 import lombok.Setter;
 import org.springframework.beans.factory.annotation.Value;
@@ -68,8 +69,26 @@ public class RqueueSchedulerConfig {
   @Value("${rqueue.scheduler.scheduled.message.time.interval:2000}")
   private long scheduledMessageTimeIntervalInMilli;
 
+  // How long the application should wait for task termination
+  @Value("${rqueue.scheduler.termination.wait.time:200}")
+  private long terminationWaitTime;
 
   // Maximum delay for message mover task due to failure
   @Value("${rqueue.scheduler.max.message.mover.delay:60000}")
   private long maxMessageMoverDelay;
+
+  // Minimum amount of time between two consecutive message move calls
+  @Value("${rqueue.scheduler.min.message.mover.delay:100}")
+  private long minMessageMoverDelay;
+
+  // Maximum number of messages that should be copied from scheduled to normal queue
+  @Value("${rqueue.scheduler.max.message.count:100}")
+  private long maxMessageCount;
+
+  public long minMessageMoveDelay() {
+    if (minMessageMoverDelay <= 0) {
+      return Constants.MIN_SCHEDULE_INTERVAL;
+    }
+    return minMessageMoverDelay;
+  }
 }