Concurrency issue when task executor is provided. (#123)

* Concurrency issue when task executor is provided.

* remove exit

* Removed unused property

* increase timeout

Author: sonus21

CHANGELOG.md

@@ -1,5 +1,9 @@
 # [Rqueue] New and Notable Changes
 
+### [2.10.1] - 18-Oct-2021
+
+* Fixes for concurrency when task executor is provided see issue #[122]
+
 ### [2.10.0] - 10-Oct-2021
 
 ### Fixes
@@ -8,13 +12,14 @@
 * Fixes message move message count (by default 1000 messages are moved)
 * Potential issue in rename collection
 * More than one (-) sign in the dashboard
-* Fixes for server context path. Rqueue end points would be served relative to x-forwarded-prefix/server.servlet.context-path
+* Fixes for server context path. Rqueue end points would be served relative to
+  x-forwarded-prefix/server.servlet.context-path
 
 ### Features
 
 * Display completed jobs in the dashboard
 * Option to choose number of days in the chart
-  ReactiveWebViewTest
+
 ### [2.9.0] - 30-Jul-2021
 
 ### Fixes
@@ -273,3 +278,7 @@ Fixes:
 [2.9.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.9.0-RELEASE
 
 [2.10.0]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.10.0-RELEASE
+
+[2.10.1]: https://repo1.maven.org/maven2/com/github/sonus21/rqueue-core/2.10.1-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.0-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring-boot-starter:2.10.1-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
         <groupId>com.github.sonus21</groupId>
         <artifactId>rqueue-spring-boot-starter</artifactId>
-        <version>2.10.0-RELEASE</version>
+        <version>2.10.1-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.0-RELEASE'
+        implementation 'com.github.sonus21:rqueue-spring:2.10.1-RELEASE'
     ```
   * Maven
     ```xml
      <dependency>
        <groupId>com.github.sonus21</groupId>
        <artifactId>rqueue-spring</artifactId>
-       <version>2.10.0-RELEASE</version>
+       <version>2.10.1-RELEASE</version>
      </dependency>
     ```
 

build.gradle

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

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

@@ -144,6 +144,7 @@
    * have the same concurrency.
    *
    * @return concurrency for this worker.
+   * @see #priority()
    */
   String concurrency() default "-1";
 
@@ -158,7 +159,7 @@
    *
    * <p>Priority can be any number. There are two priority control modes. 1. Strict 2. Weighted, in
    * strict priority mode queue with higher priority is preferred over other queues. In case of
-   * weighted a round robin approach is used, and weight is followed.
+   * weighted a round-robin approach is used, and weight is followed.
    *
    * @return the priority for this listener.
    * @see #priorityGroup()

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

@@ -113,12 +113,6 @@ public class RqueueConfig {
   @Value("${rqueue.retry.per.poll:1}")
   private int retryPerPoll;
 
-  @Value("${rqueue.add.default.queue.with.queue.level.priority:true}")
-  private boolean addDefaultQueueWithQueueLevelPriority;
-
-  @Value("${rqueue.default.queue.with.queue.level.priority:-1}")
-  private int defaultQueueWithQueueLevelPriority;
-
   @Value("${rqueue.net.proxy.host:}")
   private String proxyHost;
 

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

@@ -211,16 +211,17 @@ public Integer getMaxNumWorkers() {
    * for every queue.
    *
    * <p>When you're using custom executor then you should set this number as (thread pool max size -
-   * number of queues) given executor is not shared. The maxNumWorkers tells how many workers you
-   * want to run in parallel for all listeners, for example if you have 3 listeners, and you have
-   * set this as 10 then all 3 listeners would be running maximum **combined 10 jobs** at any point
-   * of time.
+   * number of queues) given executor is not shared with other application component. The
+   * maxNumWorkers tells how many workers you want to run in parallel for all listeners those are
+   * not having configured concurrency. For example if you have 3 queues without concurrency, and
+   * you have set this as 10 then all 3 listeners would be running maximum **combined 10 jobs** at
+   * any point of time. Queues having concurrency will be running at the configured concurrency.
    *
    * <p>What would happen if I set this to very high value while using custom executor? <br>
    * 1. Task(s) would be rejected by the executor unless queue size is non-zero <br>
    * 2. When queue size is non-zero then it can create duplicate message problem, since the polled
    * message has not been processed yet. This will happen when {@link
-   * RqueueListener#visibilityTimeout()} is smaller than the time a task took to execute from the
+   * RqueueListener#visibilityTimeout()} is smaller than the time a task takes to execute from the
    * time of polling to final execution.
    *
    * @param maxNumWorkers Maximum number of workers.

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

@@ -101,7 +101,7 @@ public QueueConfig toConfig() {
   List<QueueDetail> expandQueueDetail(boolean addDefault, int priority) {
     List<QueueDetail> queueDetails = new ArrayList<>();
     for (Entry<String, Integer> entry : getPriority().entrySet()) {
-      QueueDetail cloneQueueDetail = cloneQueueDetail(entry.getKey(), entry.getValue(), true, name);
+      QueueDetail cloneQueueDetail = cloneQueueDetail(entry.getKey(), entry.getValue(), name);
       queueDetails.add(cloneQueueDetail);
     }
     if (addDefault) {
@@ -121,7 +121,7 @@ List<QueueDetail> expandQueueDetail(boolean addDefault, int priority) {
   }
 
   private QueueDetail cloneQueueDetail(
-      String priorityName, Integer priority, boolean systemGenerated, String priorityGroup) {
+      String priorityName, Integer priority, String priorityGroup) {
     if (priority == null || priorityName == null) {
       throw new IllegalStateException("priority name is null");
     }
@@ -140,7 +140,7 @@ private QueueDetail cloneQueueDetail(
         .completedQueueName(completedQueueName + suffix)
         .active(active)
         .batchSize(batchSize)
-        .systemGenerated(systemGenerated)
+        .systemGenerated(true)
         .priorityGroup(priorityGroup)
         .concurrency(concurrency)
         .priority(Collections.singletonMap(Constants.DEFAULT_PRIORITY_KEY, priority))

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

@@ -255,11 +255,29 @@ private void initializeQueue() {
       defaultTaskExecutor = true;
       taskExecutor = createDefaultTaskExecutor(queueDetails);
     } else {
-      initializeThreadMap(queueDetails, taskExecutor, false, queueDetails.size());
+      initializeThreadMapForNonDefaultExecutor(queueDetails);
     }
     initializeRunningQueueState();
   }
 
+  private void initializeThreadMapForNonDefaultExecutor(
+      List<QueueDetail> registeredActiveQueueDetail) {
+    List<QueueDetail> queueDetails =
+        registeredActiveQueueDetail.stream()
+            .filter(e -> !e.isSystemGenerated())
+            .collect(Collectors.toList());
+    List<QueueDetail> withoutConcurrency = new ArrayList<>();
+    for (QueueDetail queueDetail : queueDetails) {
+      if (queueDetail.getConcurrency().isValid()) {
+        addExecutorForConcurrencyBasedQueue(queueDetail, taskExecutor, false);
+      } else {
+        withoutConcurrency.add(queueDetail);
+      }
+    }
+    initializeThreadMap(
+        withoutConcurrency, taskExecutor, false, getWorkersCount(withoutConcurrency.size()));
+  }
+
   private void initialize() {
     initializeQueue();
     this.postProcessingHandler =
@@ -296,9 +314,12 @@ private void initializeThreadMap(
       AsyncTaskExecutor taskExecutor,
       boolean defaultExecutor,
       int workersCount) {
+    if (queueDetails.isEmpty()) {
+      return;
+    }
+    QueueThreadPool pool = new QueueThreadPool(taskExecutor, defaultExecutor, workersCount);
     for (QueueDetail queueDetail : queueDetails) {
-      queueThreadMap.put(
-          queueDetail.getName(), new QueueThreadPool(taskExecutor, defaultExecutor, workersCount));
+      queueThreadMap.put(queueDetail.getName(), pool);
     }
   }
 
@@ -332,16 +353,19 @@ private AsyncTaskExecutor createNonConcurrencyBasedExecutor(
     return executor;
   }
 
+  private void addExecutorForConcurrencyBasedQueue(
+      QueueDetail queueDetail, AsyncTaskExecutor executor, boolean defaultTaskExecutor) {
+    int maxJobs = queueDetail.getConcurrency().getMax();
+    QueueThreadPool threadPool = new QueueThreadPool(executor, defaultTaskExecutor, maxJobs);
+    queueThreadMap.put(queueDetail.getName(), threadPool);
+  }
+
   private void createExecutor(QueueDetail queueDetail) {
     Concurrency concurrency = queueDetail.getConcurrency();
-    int queueCapacity = 0;
-    int maxJobs = concurrency.getMax();
     int corePoolSize = concurrency.getMin();
     int maxPoolSize = concurrency.getMax();
-    AsyncTaskExecutor executor =
-        createTaskExecutor(queueDetail, corePoolSize, maxPoolSize, queueCapacity);
-    QueueThreadPool threadPool = new QueueThreadPool(executor, true, maxJobs);
-    queueThreadMap.put(queueDetail.getName(), threadPool);
+    AsyncTaskExecutor executor = createTaskExecutor(queueDetail, corePoolSize, maxPoolSize);
+    addExecutorForConcurrencyBasedQueue(queueDetail, executor, true);
   }
 
   public AsyncTaskExecutor createDefaultTaskExecutor(
@@ -362,15 +386,14 @@ public AsyncTaskExecutor createDefaultTaskExecutor(
   }
 
   private AsyncTaskExecutor createTaskExecutor(
-      QueueDetail queueDetail, int corePoolSize, int maxPoolSize, int queueCapacity) {
+      QueueDetail queueDetail, int corePoolSize, int maxPoolSize) {
     String name = ThreadUtils.getWorkerName(queueDetail.getName());
-    return ThreadUtils.createTaskExecutor(
-        name, name + "-", corePoolSize, maxPoolSize, queueCapacity);
+    return ThreadUtils.createTaskExecutor(name, name + "-", corePoolSize, maxPoolSize, 0);
   }
 
   private List<QueueDetail> getQueueDetail(String queue, MappingInformation mappingInformation) {
     int numRetry = mappingInformation.getNumRetry();
-    if (!mappingInformation.getDeadLetterQueueName().isEmpty() && numRetry == -1) {
+    if (!StringUtils.isEmpty(mappingInformation.getDeadLetterQueueName()) && numRetry == -1) {
       log.warn(
           "Dead letter queue {} is set but retry is not set",
           mappingInformation.getDeadLetterQueueName());
@@ -403,12 +426,13 @@ private List<QueueDetail> getQueueDetail(String queue, MappingInformation mappin
             .priority(priority)
             .priorityGroup(priorityGroup)
             .build();
+    List<QueueDetail> queueDetails;
     if (queueDetail.getPriority().size() <= 1) {
-      return Collections.singletonList(queueDetail);
+      queueDetails = Collections.singletonList(queueDetail);
+    } else {
+      queueDetails = queueDetail.expandQueueDetail(true, -1);
     }
-    return queueDetail.expandQueueDetail(
-        rqueueConfig.isAddDefaultQueueWithQueueLevelPriority(),
-        rqueueConfig.getDefaultQueueWithQueueLevelPriority());
+    return queueDetails;
   }
 
   @Override
@@ -450,6 +474,7 @@ protected void doStart() {
 
   private Map<String, QueueThreadPool> getQueueThreadMap(
       String groupName, List<QueueDetail> queueDetails) {
+    // this happens only for queue having priorities like critical:10,high:5,low:3
     QueueThreadPool queueThreadPool = queueThreadMap.get(groupName);
     if (queueThreadPool != null) {
       return queueDetails.stream()

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

@@ -82,7 +82,7 @@ private void execute(
               queueThreadPool));
     } catch (Exception e) {
       if (e instanceof TaskRejectedException) {
-        queueThreadPool.taskRejected();
+        queueThreadPool.taskRejected(queueDetail, message);
       }
       log(Level.WARN, "Execution failed Msg: {}", e, message);
       release(postProcessingHandler, queueThreadPool, queueDetail, message);

rqueue-core/src/main/java/com/github/sonus21/rqueue/utils/QueueThreadPool.java

@@ -16,6 +16,8 @@
 
 package com.github.sonus21.rqueue.utils;
 
+import com.github.sonus21.rqueue.core.RqueueMessage;
+import com.github.sonus21.rqueue.listener.QueueDetail;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
@@ -81,7 +83,10 @@ public String destroy() {
     return null;
   }
 
-  public void taskRejected() {
-    log.warn("Task rejected by executor");
+  public void taskRejected(QueueDetail queueDetail, RqueueMessage message) {
+    log.warn(
+        "Task rejected by executor Queue: {}, Message: {}",
+        queueDetail.getName(),
+        message.getMessage());
   }
 }