Hi,
I have a simple example code for playing with Scheduler component and to see
what is blocking-where during routing:
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.scheduler.SchedulerComponent;
/**
* A Camel Java DSL Router
*/
public class MyRouteBuilder extends RouteBuilder {
/**
* Let's configure the Camel routing rules using Java code...
*/
public void configure() {
//
http://camel.465427.n5.nabble.com/Camel-scheduler-is-not-concurrent-td5781066.html
// number of concurrent task a thread pool should have
SchedulerComponent comp = getContext().getComponent("scheduler",
SchedulerComponent.class);
comp.setConcurrentTasks(2);
from("scheduler://trigger?delay=2000&scheduler.concurrentTasks=2&repeatCount=3").routeId("scheduler")
.log("1")
.inOut("seda:route1")
.log("1.1");
from("seda:route1?concurrentConsumers=1").routeId("first route")
.log("2")
.delay(5000)
.log("2.1")
.inOut("seda:route2")
.log("2.2");
from("seda:route2").routeId("second route")
.log("3")
.delay(3000)
.log("3.1");
}
}
Expected behavior is
- scheduler fires an event
- event goes to route 1, thread got blocked
- scheduler fires again, event goes to route 1 then blocked to
- etc.
Instead of that I get this output:
19:59:49,895 [hread #0 - scheduler://trigger] scheduler
INFO 1
19:59:49,895 [hread #1 - scheduler://trigger] scheduler
INFO 1
19:59:49,942 [l-1) thread #2 - seda://route1] first route
INFO 2
19:59:49,945 [l-1) thread #2 - seda://route1] first route
INFO 2
19:59:54,946 [el (camel-1) thread #5 - Delay] first route
INFO 2.1
19:59:54,947 [el (camel-1) thread #6 - Delay] first route
INFO 2.1
19:59:54,948 [l-1) thread #3 - seda://route2] second route
INFO 3
19:59:54,950 [l-1) thread #3 - seda://route2] second route
INFO 3
19:59:57,950 [el (camel-1) thread #7 - Delay] second route
INFO 3.1
19:59:57,951 [el (camel-1) thread #5 - Delay] first route
INFO 2.2
19:59:57,951 [el (camel-1) thread #7 - Delay] second route
INFO 3.1
19:59:57,951 [hread #0 - scheduler://trigger] scheduler
INFO 1.1
19:59:57,952 [el (camel-1) thread #6 - Delay] first route
INFO 2.2
19:59:57,952 [hread #1 - scheduler://trigger] scheduler
INFO 1.1
19:59:59,954 [hread #0 - scheduler://trigger] scheduler
INFO 1
19:59:59,955 [l-1) thread #2 - seda://route1] first route
INFO 2
19:59:59,958 [hread #0 - scheduler://trigger] DefaultErrorHandler
ERROR Failed delivery for (MessageId: D72AAD6A473D1EB-0000000000000004 on
ExchangeId: D72AAD6A473D1EB-0000000000000004). Exhausted after delivery
attempt: 1 caught: org.apache.camel.ExchangeTimedOutException: The OUT message
was not received within: 30000 millis.
Exchange[D72AAD6A473D1EB-0000000000000004]
Message History (complete message history is disabled)
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor
Elapsed (ms)
[scheduler ] [scheduler ]
[from[scheduler://trigger?delay=2000&repeatCount=3&scheduler.concurrentTasks=2]]
[ 4]
...
[scheduler ] [to1 ] [seda:route1
] [ 0]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.ExchangeTimedOutException: The OUT message was not received
within: 30000 millis. Exchange[D72AAD6A473D1EB-0000000000000004]
at
org.apache.camel.component.seda.SedaProducer.process(SedaProducer.java:129)
~[camel-seda-3.8.0.jar:3.8.0]
at
org.apache.camel.processor.SendProcessor.process(SendProcessor.java:169)
~[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:395)
[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
[camel-base-engine-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
[camel-base-engine-3.8.0.jar:3.8.0]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:312)
[camel-base-engine-3.8.0.jar:3.8.0]
at
org.apache.camel.component.scheduler.SchedulerConsumer.sendTimerExchange(SchedulerConsumer.java:58)
[camel-scheduler-3.8.0.jar:3.8.0]
at
org.apache.camel.component.scheduler.SchedulerConsumer.poll(SchedulerConsumer.java:43)
[camel-scheduler-3.8.0.jar:3.8.0]
at
org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:190)
[camel-support-3.8.0.jar:3.8.0]
at
org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107)
[camel-support-3.8.0.jar:3.8.0]
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
[?:?]
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?]
at java.lang.Thread.run(Thread.java:834) [?:?]
19:59:59,988 [hread #0 - scheduler://trigger] SchedulerConsumer
WARN Error processing exchange. Exchange[D72AAD6A473D1EB-0000000000000004].
Caused by: [org.apache.camel.ExchangeTimedOutException - The OUT message was
not received within: 30000 millis. Exchange[D72AAD6A473D1EB-0000000000000004]]
org.apache.camel.ExchangeTimedOutException: The OUT message was not received
within: 30000 millis. Exchange[D72AAD6A473D1EB-0000000000000004]
at
org.apache.camel.component.seda.SedaProducer.process(SedaProducer.java:129)
~[camel-seda-3.8.0.jar:3.8.0]
at
org.apache.camel.processor.SendProcessor.process(SendProcessor.java:169)
~[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:395)
~[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148)
[camel-base-engine-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:60)
[camel-base-engine-3.8.0.jar:3.8.0]
at org.apache.camel.processor.Pipeline.process(Pipeline.java:147)
[camel-core-processor-3.8.0.jar:3.8.0]
at
org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:312)
[camel-base-engine-3.8.0.jar:3.8.0]
at
org.apache.camel.component.scheduler.SchedulerConsumer.sendTimerExchange(SchedulerConsumer.java:58)
[camel-scheduler-3.8.0.jar:3.8.0]
at
org.apache.camel.component.scheduler.SchedulerConsumer.poll(SchedulerConsumer.java:43)
[camel-scheduler-3.8.0.jar:3.8.0]
at
org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:190)
[camel-support-3.8.0.jar:3.8.0]
at
org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107)
[camel-support-3.8.0.jar:3.8.0]
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
[?:?]
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?]
at java.lang.Thread.run(Thread.java:834) [?:?]
20:00:04,956 [el (camel-1) thread #5 - Delay] first route
INFO 2.1
20:00:04,957 [l-1) thread #3 - seda://route2] second route
INFO 3
20:00:07,959 [el (camel-1) thread #8 - Delay] second route
INFO 3.1
20:00:07,960 [el (camel-1) thread #5 - Delay] first route
INFO 2.2
Beside events are not triggered and routed as expected - most probably due to
scheduler.concurrentTasks=2 in scheduler route definition (unintentional
leftover before reading
http://camel.465427.n5.nabble.com/Camel-scheduler-is-not-concurrent-td5781066.html)
- the exceptions buzzing around 30secs timeouts were really strange...
What do you think?
Br,
Zoltan
