[camel] branch master updated: CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed to poolSize and avoid scheduler to cause concurrent triggers that causes routing problems.

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/master by this push:
     new 07c60df  CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed 
to poolSize and avoid scheduler to cause concurrent triggers that causes 
routing problems.
07c60df is described below

commit 07c60df69280cfe282148d146d8c2775d43cd9b3
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Mon Mar 22 16:57:48 2021 +0100

    CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed to poolSize 
and avoid scheduler to cause concurrent triggers that causes routing problems.
---
 .../camel/component/aws2/sqs/Sqs2Endpoint.java      |  2 +-
 .../camel/component/ironmq/IronMQEndpoint.java      |  2 +-
 .../DefaultScheduledPollConsumerScheduler.java      | 21 +++++++++++++++++----
 3 files changed, 19 insertions(+), 6 deletions(-)

diff --git 
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
 
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
index 1eb4014..40451c0 100644
--- 
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
+++ 
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
@@ -108,7 +108,7 @@ public class Sqs2Endpoint extends ScheduledPollEndpoint 
implements HeaderFilterS
         scheduler.setUseFixedDelay(sqsConsumer.isUseFixedDelay());
         scheduler.setInitialDelay(sqsConsumer.getInitialDelay());
         scheduler.setTimeUnit(sqsConsumer.getTimeUnit());
-        scheduler.setConcurrentTasks(configuration.getConcurrentConsumers());
+        
scheduler.setConcurrentConsumers(configuration.getConcurrentConsumers());
         sqsConsumer.setScheduler(scheduler);
         return sqsConsumer;
     }
diff --git 
a/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
 
b/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
index 4e8c116..1eda2fc 100644
--- 
a/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
+++ 
b/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
@@ -66,7 +66,7 @@ public class IronMQEndpoint extends ScheduledPollEndpoint {
         scheduler.setUseFixedDelay(ironMQConsumer.isUseFixedDelay());
         scheduler.setInitialDelay(ironMQConsumer.getInitialDelay());
         scheduler.setTimeUnit(ironMQConsumer.getTimeUnit());
-        scheduler.setConcurrentTasks(configuration.getConcurrentConsumers());
+        
scheduler.setConcurrentConsumers(configuration.getConcurrentConsumers());
         ironMQConsumer.setScheduler(scheduler);
         return ironMQConsumer;
     }
diff --git 
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
 
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
index af6aa81..3d77a6a 100644
--- 
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
+++ 
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
@@ -48,6 +48,7 @@ public class DefaultScheduledPollConsumerScheduler extends 
ServiceSupport implem
     private volatile List<ScheduledFuture<?>> futures = new ArrayList<>();
     private Runnable task;
     private int poolSize = 1;
+    private int concurrentConsumers = 1;
 
     private long initialDelay = -1;
     private long delay = -1;
@@ -111,6 +112,14 @@ public class DefaultScheduledPollConsumerScheduler extends 
ServiceSupport implem
         this.scheduledExecutorService = scheduledExecutorService;
     }
 
+    public int getConcurrentConsumers() {
+        return concurrentConsumers;
+    }
+
+    public void setConcurrentConsumers(int concurrentConsumers) {
+        this.concurrentConsumers = concurrentConsumers;
+    }
+
     public int getPoolSize() {
         return poolSize;
     }
@@ -170,16 +179,20 @@ public class DefaultScheduledPollConsumerScheduler 
extends ServiceSupport implem
                             currentInitialDelay, currentDelay, 
getTimeUnit().name().toLowerCase(Locale.ENGLISH),
                             consumer.getEndpoint());
                 }
-                
futures.add(scheduledExecutorService.scheduleWithFixedDelay(task, 
currentInitialDelay, currentDelay,
-                        getTimeUnit()));
+                for (int i = 0; i < concurrentConsumers; i++) {
+                    
futures.add(scheduledExecutorService.scheduleWithFixedDelay(task, 
currentInitialDelay, currentDelay,
+                            getTimeUnit()));
+                }
             } else {
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("Scheduling poll (fixed rate) with initialDelay: 
{}, delay: {} ({}) for: {}",
                             currentInitialDelay, currentDelay, 
getTimeUnit().name().toLowerCase(Locale.ENGLISH),
                             consumer.getEndpoint());
                 }
-                futures.add(scheduledExecutorService.scheduleAtFixedRate(task, 
currentInitialDelay, currentDelay,
-                        getTimeUnit()));
+                for (int i = 0; i < concurrentConsumers; i++) {
+                    
futures.add(scheduledExecutorService.scheduleAtFixedRate(task, 
currentInitialDelay, currentDelay,
+                            getTimeUnit()));
+                }
             }
         }
     }