[camel] branch master updated: CAMEL-14442: Fixed camel-scheduler to use same executor service for same scheduler name.

[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 3d441f2  CAMEL-14442: Fixed camel-scheduler to use same executor 
service for same scheduler name.
3d441f2 is described below

commit 3d441f2cd2bf471b361a245324ccfb24c3582e3b
Author: Claus Ibsen <claus.ib...@gmail.com>
AuthorDate: Tue Jan 28 08:51:24 2020 +0100

    CAMEL-14442: Fixed camel-scheduler to use same executor service for same 
scheduler name.
---
 .../component/scheduler/TwoSchedulerTest.java      | 13 ++++++++++--
 .../camel/support/ScheduledPollConsumer.java       | 23 +++++++++++-----------
 2 files changed, 22 insertions(+), 14 deletions(-)

diff --git 
a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerTest.java
 
b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerTest.java
index dc8a591..44e9520 100644
--- 
a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerTest.java
+++ 
b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerTest.java
@@ -28,15 +28,24 @@ public class TwoSchedulerTest extends ContextTestSupport {
         getMockEndpoint("mock:b").expectedMinimumMessageCount(2);
 
         assertMockEndpointsSatisfied();
+
+        // should use same thread as they share the same scheduler
+        String tn1 = 
getMockEndpoint("mock:a").getReceivedExchanges().get(0).getMessage().getHeader("tn",
 String.class);
+        String tn2 = 
getMockEndpoint("mock:b").getReceivedExchanges().get(0).getMessage().getHeader("tn",
 String.class);
+        assertSame(tn1, tn2);
     }
 
     @Override
     protected RouteBuilder createRouteBuilder() throws Exception {
         return new RouteBuilder() {
             public void configure() {
-                from("scheduler://foo?delay=100").to("mock:a");
+                from("scheduler://foo?delay=100")
+                        .setHeader("tn", simple("${threadName}"))
+                        .to("mock:a");
 
-                from("scheduler://foo?delay=200").to("mock:b");
+                from("scheduler://foo?delay=200")
+                        .setHeader("tn", simple("${threadName}"))
+                        .to("mock:b");
             }
         };
     }
diff --git 
a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
 
b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
index 4414ede..0b8efcc 100644
--- 
a/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
+++ 
b/core/camel-support/src/main/java/org/apache/camel/support/ScheduledPollConsumer.java
@@ -435,17 +435,6 @@ public abstract class ScheduledPollConsumer extends 
DefaultConsumer implements R
             LOG.debug("Using backoff[multiplier={}, idleThreshold={}, 
errorThreshold={}] on {}", backoffMultiplier, backoffIdleThreshold, 
backoffErrorThreshold, getEndpoint());
         }
 
-        if (scheduler == null) {
-            DefaultScheduledPollConsumerScheduler scheduler = new 
DefaultScheduledPollConsumerScheduler(scheduledExecutorService);
-            scheduler.setDelay(delay);
-            scheduler.setInitialDelay(initialDelay);
-            scheduler.setTimeUnit(timeUnit);
-            scheduler.setUseFixedDelay(useFixedDelay);
-            this.scheduler = scheduler;
-        }
-        scheduler.setCamelContext(getEndpoint().getCamelContext());
-        scheduler.onInit(this);
-
         // configure scheduler with options from this consumer
         if (schedulerProperties != null && !schedulerProperties.isEmpty()) {
             // need to use a copy in case the consumer is restarted so we keep 
the properties
@@ -459,7 +448,6 @@ public abstract class ScheduledPollConsumer extends 
DefaultConsumer implements R
             }
         }
 
-        ObjectHelper.notNull(scheduler, "scheduler", this);
         ObjectHelper.notNull(pollStrategy, "pollStrategy", this);
     }
 
@@ -467,6 +455,17 @@ public abstract class ScheduledPollConsumer extends 
DefaultConsumer implements R
     protected void doStart() throws Exception {
         super.doStart();
 
+        if (scheduler == null) {
+            DefaultScheduledPollConsumerScheduler scheduler = new 
DefaultScheduledPollConsumerScheduler(scheduledExecutorService);
+            scheduler.setDelay(delay);
+            scheduler.setInitialDelay(initialDelay);
+            scheduler.setTimeUnit(timeUnit);
+            scheduler.setUseFixedDelay(useFixedDelay);
+            this.scheduler = scheduler;
+        }
+        scheduler.setCamelContext(getEndpoint().getCamelContext());
+        scheduler.onInit(this);
+
         if (scheduler != null) {
             scheduler.scheduleTask(this);
             ServiceHelper.startService(scheduler);