Multiple Fixes (#181)

* do not schedule a message if the job is terminated
* update local metadata to handle the parallel deletion
* Task rejected when threads are not available, sleep instead of poll
* Display DatTime string in local time

Author: sonus21

.circleci/config.yml

@@ -117,7 +117,8 @@ aliases:
 executors:
   rqueue-executor:
     machine:
-      image: ubuntu-2004:202201-02
+      image: ubuntu-2004:202010-01
+    resource_class: large
 
     working_directory: ~/repo
 
@@ -267,15 +268,15 @@ workflows:
       - integration_test:
           requires:
             - producer_only_test
-      - reactive_integration_test:
+      - redis_custer_test:
           requires:
             - integration_test
-      - redis_custer_test:
+      - reactive_integration_test:
           requires:
-            - reactive_integration_test
+            - redis_custer_test
       - report_code_coverage:
           requires:
-            - redis_custer_test
+            - reactive_integration_test
   main_v0:
     jobs:
       - build_v0

CHANGELOG.md

@@ -1,5 +1,11 @@
 # [Rqueue] New and Notable Changes
 
+### [2.13.0] - 25-Dec-2022
+### Fixes
+* 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
 * Upgraded Pebble version for CVE
@@ -319,4 +325,6 @@ Fixes:
 
 [2.12.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.12.0-RELEASE
 
+[2.13.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.13.0-RELEASE
+
 [122]: https://github.com/sonus21/rqueue/issues/122

README.md

@@ -53,6 +53,7 @@ well, where all services code is in Spring.
 ### Requirements
 
 * Spring 5+
+* Java 1.8+
 * Spring boot 2+
 * Lettuce client for Redis cluster
 * Read master preference for Redis cluster
@@ -64,41 +65,44 @@ well, where all services code is in Spring.
 Snapshot Version: [https://s01.oss.sonatype.org/content/repositories/snapshots/com/github/sonus21/](https://s01.oss.sonatype.org/content/repositories/snapshots/com/github/sonus21/)
 Release Version: [Maven central](https://search.maven.org/search?q=g:com.github.sonus21)
 
-#### Spring-boot
+---
+#### Spring Boot
 
 * Get the latest one
   from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AND%20a:rqueue-spring-boot-starter)
 * Add dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.11-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.13.0-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
         <groupId>com.github.sonus21</groupId>
         <artifactId>rqueue-spring-boot-starter</artifactId>
-        <version>2.12.0-RELEASE</version>
+        <version>2.13.0-RELEASE</version>
     </dependency>
     ```
 
   No additional configurations are required, only dependency is required.
 
+---
+
 #### Spring framework
 
 * Get the latest one
   from [Maven central](https://search.maven.org/search?q=g:com.github.sonus21%20AND%20a:rqueue-spring)
 * Add Dependency
   * Gradle
     ```groovy
-        implementation 'com.github.sonus21:rqueue-spring:2.12.0-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring:2.13.0-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
        <groupId>com.github.sonus21</groupId>
        <artifactId>rqueue-spring</artifactId>
-       <version>2.12.0-RELEASE</version>
+       <version>2.13.0-RELEASE</version>
      </dependency>
     ```
 
@@ -119,11 +123,12 @@ public class Application {
 }
 ```
 
+---
 ### Message publishing/Task submission
 
 All messages need to be sent using `RqueueMessageEnqueuer` bean's `enqueueXXX`, `enqueueInXXX`
 and `enqueueAtXXX` methods. It has handful number of `enqueue`, `enqueueIn`, `enqueueAt` methods, we
-can use one of them based on the use case.
+can use any one of them based on the use case.
 
 ```java
 public class MessageService {
@@ -162,14 +167,16 @@ public class MessageService {
     rqueueMessageEnqueuer.enqueueWithPriority("sms-queue", priority.value(), sms);
   }
 
-  // enqueue periodic job, email should be sent every 30 seconds
+  // Index chat every 1 minute
   public void sendPeriodicEmail(Email email) {
-    rqueueMessageEnqueuer.enqueuePeriodic("email-queue", invoice, 30_000);
+    rqueueMessageEnqueuer.enqueuePeriodic("chat-indexer", chatIndexer, 60_000);
   }
 
 }
 ```
 
+---
+
 ### Worker/Consumer/Task Executor/Listener
 
 Any method that's part of spring bean, can be marked as worker/message listener
@@ -223,19 +230,21 @@ public class MessageListener {
 }
 ```
 
-## Queue Statistics
+---
+## Dashboard
 
-Micrometer based dashboard for queue
+Link: [http://localhost:8080/rqueue](http://localhost:8080/rqueue)
 
-[![Grafana Dashboard](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/grafana-dashboard.png)](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/grafana-dashboard.png)
 
-## Web
+[![Dashboard](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/stats-graph.png)](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/stats-graph.png)
 
-Link: [http://localhost:8080/rqueue](http://localhost:8080/rqueue)
 
-#### Dashboard
+#### Queue Statistics
+
+Micrometer based dashboard for queue
+
+[![Grafana Dashboard](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/grafana-dashboard.png)](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/grafana-dashboard.png)
 
-[![Dashboard](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/stats-graph.png)](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/stats-graph.png)
 
 #### Message Waiting For Execution
 
@@ -245,6 +254,8 @@ Link: [http://localhost:8080/rqueue](http://localhost:8080/rqueue)
 
 [![Jobs](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/jobs.png)](https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/jobs.png)
 
+---
+
 ## Status
 
 Rqueue is stable and production ready, it's processing 100K+ messages daily in production
@@ -264,6 +275,8 @@ PR/[issue](https://github.com/sonus21/rqueue/issues/new?template=i-m-using-rqueu
 <a href="https://www.chaotiinfo.cn"><img src="https://raw.githubusercontent.com/sonus21/rqueue/master/docs/static/users/chaoti-info.png" alt="CHAOTI INFO TECH(SHENZHEN)" height="60" align="middle"/></a>
 &nbsp;&nbsp;
 
+---
+
 <!---- Signing Key
 ~/.gradle/gradle.properties file
 
@@ -286,7 +299,7 @@ signing.secretKeyRingFile=/Users/sonu/.gnupg/secring.gpg generate this as `gpg -
 
 * Please report bug,question,feature(s)
   to [issue](https://github.com/sonus21/rqueue/issues/new/choose) tracker.
-* Ask question on StackOverflow using [rqueue](https://stackoverflow.com/tags/rqueue) tag
+* Ask question on StackOverflow using [#rqueue](https://stackoverflow.com/tags/rqueue) tag
 
 ## Contribution
 

build.gradle

@@ -3,7 +3,9 @@ plugins {
     id 'com.github.kt3k.coveralls' version '2.6.3'
     id 'nebula.optional-base' version '5.0.2'
     id "com.adarshr.test-logger" version "2.1.1"
+    id 'org.gradle.test-retry' version '1.0.0'
 }
+
 allprojects {
     apply plugin: 'idea'
     apply plugin: 'java'
@@ -13,6 +15,7 @@ allprojects {
     apply plugin: 'jacoco'
     apply plugin: 'nebula.optional-base'
     apply plugin: 'com.adarshr.test-logger'
+    apply plugin: 'org.gradle.test-retry'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8
@@ -21,6 +24,7 @@ allprojects {
         mavenCentral()
     }
 }
+
 ext {
     springBootVersion = System.getenv("SPRING_BOOT_VERSION")
     springVersion = System.getenv("SPRING_VERSION")
@@ -70,7 +74,7 @@ ext {
 
 subprojects {
     group = 'com.github.sonus21'
-    version = '2.12.0-RELEASE'
+    version = '2.13.0-RELEASE'
 
     dependencies {
         // https://mvnrepository.com/artifact/org.springframework/spring-messaging
@@ -137,6 +141,7 @@ task codeCoverageReport(type: JacocoReport, group: 'verification', description:
                             'com/github/sonus21/rqueue/models/response',
                             'com/github/sonus21/rqueue/core/RqueueMessageSender*',
                             'com/github/sonus21/rqueue/utils/PrefixLogger*',
+                            'com/github/sonus21/rqueue/utils/StackTraceUtil*',
                             'com/github/sonus21/rqueue/core/ScheduledTaskDetail*',
                     ]
             )

gradle/test-runner.gradle

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 Sonu Kumar
+ * Copyright 2022 Sonu Kumar
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -38,4 +38,10 @@ test {
         html.enabled = System.getenv("CIRCLECI") != "true"
         html.destination = file("$buildDir/reports/junit/html")
     }
+
+    retry {
+        failOnPassedAfterRetry = false
+        maxFailures = 10
+        maxRetries = 1
+    }
 }

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

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2021 Sonu Kumar
+ *  Copyright 2022 Sonu Kumar
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -331,4 +331,15 @@ public String getLibVersion() {
   public boolean isProducer() {
     return RqueueMode.PRODUCER.equals(getMode());
   }
+
+  public Duration getMessageDurability(Long messageLife) {
+    if (messageLife == null || messageLife.intValue() == 0) {
+      return Duration.ofMinutes(messageDurabilityInMinute);
+    }
+    Duration duration = Duration.ofMillis(2 * messageLife);
+    if (duration.toMinutes() > messageDurabilityInMinute) {
+      return duration;
+    }
+    return Duration.ofMinutes(messageDurabilityInMinute);
+  }
 }

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

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2021 Sonu Kumar
+ *  Copyright 2022 Sonu Kumar
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.

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

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2021 Sonu Kumar
+ *  Copyright 2022 Sonu Kumar
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -27,7 +27,9 @@
 import lombok.ToString;
 import org.springframework.messaging.MessageHeaders;
 
-/** Internal message for Rqueue */
+/**
+ * Internal message for Rqueue
+ */
 @Getter
 @Setter
 @ToString
@@ -43,13 +45,14 @@ public class RqueueMessage extends SerializableBase {
   // Queue name on which message was enqueued
   private String queueName;
   /**
-   * JSON encoded message, this message can be converted to actual object with the help of {@link
-   * com.github.sonus21.rqueue.core.support.RqueueMessageUtils#convertMessageToObject} method
+   * JSON encoded message, this message can be converted to actual object with the help of
+   * {@link com.github.sonus21.rqueue.core.support.RqueueMessageUtils#convertMessageToObject}
+   * method
    */
   private String message;
   // Any retry count used while enqueueing
   private Integer retryCount;
-  // when this message was enqueued, this is in nano second
+  // when this message was enqueued, this is in nanosecond
   private long queuedTime;
   // when this message was supposed to be processed
   private long processAt;
@@ -66,7 +69,9 @@ public class RqueueMessage extends SerializableBase {
   // period of this task, if this is a periodic task.
   private long period;
 
-  @ToString.Exclude @JsonIgnore private MessageHeaders messageHeaders;
+  @ToString.Exclude
+  @JsonIgnore
+  private MessageHeaders messageHeaders;
 
   @JsonIgnore
   public RqueueMessage updateReEnqueuedAt() {
@@ -87,14 +92,14 @@ public boolean equals(Object other) {
 
   @JsonIgnore
   public long nextProcessAt() {
-    if (isPeriodicTask()) {
+    if (isPeriodic()) {
       return processAt + period;
     }
     throw new IllegalStateException("Only applicable for periodic message");
   }
 
   @JsonIgnore
-  public boolean isPeriodicTask() {
+  public boolean isPeriodic() {
     return period > 0;
   }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -1,5 +1,5 @@
 /*
- *  Copyright 2021 Sonu Kumar
+ *  Copyright 2022 Sonu Kumar
  *
  *  Licensed under the Apache License, Version 2.0 (the "License");
  *  you may not use this file except in compliance with the License.
@@ -68,16 +68,7 @@ abstract class BaseMessageSender {
   protected Object storeMessageMetadata(
       RqueueMessage rqueueMessage, Long delayInMillis, boolean reactive) {
     MessageMetadata messageMetadata = new MessageMetadata(rqueueMessage, MessageStatus.ENQUEUED);
-    Duration duration;
-    if (delayInMillis != null) {
-      duration = Duration.ofMillis(2 * delayInMillis);
-      long minutes = duration.toMinutes();
-      if (minutes < rqueueConfig.getMessageDurabilityInMinute()) {
-        duration = Duration.ofMinutes(rqueueConfig.getMessageDurabilityInMinute());
-      }
-    } else {
-      duration = Duration.ofMinutes(rqueueConfig.getMessageDurabilityInMinute());
-    }
+    Duration duration = rqueueConfig.getMessageDurability(delayInMillis);
     if (reactive) {
       return rqueueMessageMetadataService.saveReactive(messageMetadata, duration);
     } else {