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
commit daa4348271e84658cb19eb9341c5759da3c94296 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Mar 22 16:03:17 2021 +0100 CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed to poolSize and avoid scheduler to cause concurrent triggers that causes routing problems. --- .../camel/catalog/docs/scheduler-component.adoc | 4 +- .../scheduler/SchedulerComponentConfigurer.java | 12 ++-- .../scheduler/SchedulerEndpointConfigurer.java | 12 ++-- .../scheduler/SchedulerEndpointUriFactory.java | 2 +- .../camel/component/scheduler/scheduler.json | 4 +- .../src/main/docs/scheduler-component.adoc | 4 +- .../component/scheduler/SchedulerComponent.java | 18 +++--- .../component/scheduler/SchedulerEndpoint.java | 12 ++-- .../dsl/SchedulerComponentBuilderFactory.java | 12 ++-- .../component/scheduler/SchedulerBlockingTest.java | 44 +++++++++++++++ .../scheduler/SchedulerNoPolledMessagesTest.java | 2 +- .../TwoSchedulerConcurrentTasksOneRouteTest.java | 8 +-- .../scheduler/TwoSchedulerConcurrentTasksTest.java | 2 +- .../dsl/SchedulerEndpointBuilderFactory.java | 64 +++++++++++----------- .../DefaultScheduledPollConsumerScheduler.java | 24 ++++---- .../modules/ROOT/pages/scheduler-component.adoc | 4 +- .../ROOT/pages/camel-3x-upgrade-guide-3_10.adoc | 19 +++++++ .../modules/ROOT/pages/camel-3x-upgrade-guide.adoc | 1 + 18 files changed, 153 insertions(+), 95 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/scheduler-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/scheduler-component.adoc index 96148bd..1a571f0 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/scheduler-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/scheduler-component.adoc @@ -51,7 +51,7 @@ The Scheduler component supports 3 options, which are listed below. | Name | Description | Default | Type | *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean | *autowiredEnabled* (advanced) | Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc. | true | boolean -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int |=== // component options: END @@ -90,10 +90,10 @@ with the following path and query parameters: | *backoffErrorThreshold* (scheduler) | The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in. | | int | *backoffIdleThreshold* (scheduler) | The number of subsequent idle polls that should happen before the backoffMultipler should kick-in. | | int | *backoffMultiplier* (scheduler) | To let the scheduled polling consumer backoff if there has been a number of subsequent idles/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening again. When this option is in use then backoffIdleThreshold and/or backoffErrorThreshold must also be configured. | | int -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int | *delay* (scheduler) | Milliseconds before the next poll. | 500 | long | *greedy* (scheduler) | If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages. | false | boolean | *initialDelay* (scheduler) | Milliseconds before the first poll starts. | 1000 | long +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int | *repeatCount* (scheduler) | Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever. | 0 | long | *runLoggingLevel* (scheduler) | The consumer logs a start/complete log line when it polls. This option allows you to configure the logging level for that. There are 6 enums and the value can be one of: TRACE, DEBUG, INFO, WARN, ERROR, OFF | TRACE | LoggingLevel | *scheduledExecutorService* (scheduler) | Allows for configuring a custom/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool. | | ScheduledExecutorService diff --git a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerComponentConfigurer.java b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerComponentConfigurer.java index c4bb132..46de04c 100644 --- a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerComponentConfigurer.java +++ b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerComponentConfigurer.java @@ -25,8 +25,8 @@ public class SchedulerComponentConfigurer extends PropertyConfigurerSupport impl case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true; case "bridgeerrorhandler": case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true; - case "concurrenttasks": - case "concurrentTasks": target.setConcurrentTasks(property(camelContext, int.class, value)); return true; + case "poolsize": + case "poolSize": target.setPoolSize(property(camelContext, int.class, value)); return true; default: return false; } } @@ -38,8 +38,8 @@ public class SchedulerComponentConfigurer extends PropertyConfigurerSupport impl case "autowiredEnabled": return boolean.class; case "bridgeerrorhandler": case "bridgeErrorHandler": return boolean.class; - case "concurrenttasks": - case "concurrentTasks": return int.class; + case "poolsize": + case "poolSize": return int.class; default: return null; } } @@ -52,8 +52,8 @@ public class SchedulerComponentConfigurer extends PropertyConfigurerSupport impl case "autowiredEnabled": return target.isAutowiredEnabled(); case "bridgeerrorhandler": case "bridgeErrorHandler": return target.isBridgeErrorHandler(); - case "concurrenttasks": - case "concurrentTasks": return target.getConcurrentTasks(); + case "poolsize": + case "poolSize": return target.getPoolSize(); default: return null; } } diff --git a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointConfigurer.java b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointConfigurer.java index e1ffdd9..6e91f49 100644 --- a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointConfigurer.java +++ b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointConfigurer.java @@ -29,8 +29,6 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "backoffMultiplier": target.setBackoffMultiplier(property(camelContext, int.class, value)); return true; case "bridgeerrorhandler": case "bridgeErrorHandler": target.setBridgeErrorHandler(property(camelContext, boolean.class, value)); return true; - case "concurrenttasks": - case "concurrentTasks": target.setConcurrentTasks(property(camelContext, int.class, value)); return true; case "delay": target.setDelay(property(camelContext, long.class, value)); return true; case "exceptionhandler": case "exceptionHandler": target.setExceptionHandler(property(camelContext, org.apache.camel.spi.ExceptionHandler.class, value)); return true; @@ -41,6 +39,8 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "initialDelay": target.setInitialDelay(property(camelContext, long.class, value)); return true; case "pollstrategy": case "pollStrategy": target.setPollStrategy(property(camelContext, org.apache.camel.spi.PollingConsumerPollStrategy.class, value)); return true; + case "poolsize": + case "poolSize": target.setPoolSize(property(camelContext, int.class, value)); return true; case "repeatcount": case "repeatCount": target.setRepeatCount(property(camelContext, long.class, value)); return true; case "runlogginglevel": @@ -74,8 +74,6 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "backoffMultiplier": return int.class; case "bridgeerrorhandler": case "bridgeErrorHandler": return boolean.class; - case "concurrenttasks": - case "concurrentTasks": return int.class; case "delay": return long.class; case "exceptionhandler": case "exceptionHandler": return org.apache.camel.spi.ExceptionHandler.class; @@ -86,6 +84,8 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "initialDelay": return long.class; case "pollstrategy": case "pollStrategy": return org.apache.camel.spi.PollingConsumerPollStrategy.class; + case "poolsize": + case "poolSize": return int.class; case "repeatcount": case "repeatCount": return long.class; case "runlogginglevel": @@ -120,8 +120,6 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "backoffMultiplier": return target.getBackoffMultiplier(); case "bridgeerrorhandler": case "bridgeErrorHandler": return target.isBridgeErrorHandler(); - case "concurrenttasks": - case "concurrentTasks": return target.getConcurrentTasks(); case "delay": return target.getDelay(); case "exceptionhandler": case "exceptionHandler": return target.getExceptionHandler(); @@ -132,6 +130,8 @@ public class SchedulerEndpointConfigurer extends PropertyConfigurerSupport imple case "initialDelay": return target.getInitialDelay(); case "pollstrategy": case "pollStrategy": return target.getPollStrategy(); + case "poolsize": + case "poolSize": return target.getPoolSize(); case "repeatcount": case "repeatCount": return target.getRepeatCount(); case "runlogginglevel": diff --git a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointUriFactory.java b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointUriFactory.java index 073c2bf..8bb4b9d 100644 --- a/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointUriFactory.java +++ b/components/camel-scheduler/src/generated/java/org/apache/camel/component/scheduler/SchedulerEndpointUriFactory.java @@ -24,10 +24,10 @@ public class SchedulerEndpointUriFactory extends org.apache.camel.support.compon props.add("backoffMultiplier"); props.add("synchronous"); props.add("sendEmptyMessageWhenIdle"); + props.add("poolSize"); props.add("schedulerProperties"); props.add("exchangePattern"); props.add("initialDelay"); - props.add("concurrentTasks"); props.add("backoffIdleThreshold"); props.add("scheduler"); props.add("bridgeErrorHandler"); diff --git a/components/camel-scheduler/src/generated/resources/org/apache/camel/component/scheduler/scheduler.json b/components/camel-scheduler/src/generated/resources/org/apache/camel/component/scheduler/scheduler.json index 1918d99..ee37b47 100644 --- a/components/camel-scheduler/src/generated/resources/org/apache/camel/component/scheduler/scheduler.json +++ b/components/camel-scheduler/src/generated/resources/org/apache/camel/component/scheduler/scheduler.json @@ -24,7 +24,7 @@ "componentProperties": { "bridgeErrorHandler": { "kind": "property", "displayName": "Bridge Error Handler", "group": "consumer", "label": "consumer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a me [...] "autowiredEnabled": { "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which t [...] - "concurrentTasks": { "kind": "property", "displayName": "Concurrent Tasks", "group": "scheduler", "label": "scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of threads used by the scheduling thread pool. Is by default using a single thread" } + "poolSize": { "kind": "property", "displayName": "Pool Size", "group": "scheduler", "label": "scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread" } }, "properties": { "name": { "kind": "path", "displayName": "Name", "group": "consumer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The name of the scheduler" }, @@ -37,10 +37,10 @@ "backoffErrorThreshold": { "kind": "parameter", "displayName": "Backoff Error Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in." }, "backoffIdleThreshold": { "kind": "parameter", "displayName": "Backoff Idle Threshold", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "The number of subsequent idle polls that should happen before the backoffMultipler should kick-in." }, "backoffMultiplier": { "kind": "parameter", "displayName": "Backoff Multiplier", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "description": "To let the scheduled polling consumer backoff if there has been a number of subsequent idles\/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening agai [...] - "concurrentTasks": { "kind": "parameter", "displayName": "Concurrent Tasks", "group": "scheduler", "label": "scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of threads used by the scheduling thread pool. Is by default using a single thread" }, "delay": { "kind": "parameter", "displayName": "Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 500, "description": "Milliseconds before the next poll." }, "greedy": { "kind": "parameter", "displayName": "Greedy", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages." }, "initialDelay": { "kind": "parameter", "displayName": "Initial Delay", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1000, "description": "Milliseconds before the first poll starts." }, + "poolSize": { "kind": "parameter", "displayName": "Pool Size", "group": "scheduler", "label": "scheduler", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "description": "Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread" }, "repeatCount": { "kind": "parameter", "displayName": "Repeat Count", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "description": "Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever." }, "runLoggingLevel": { "kind": "parameter", "displayName": "Run Logging Level", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "org.apache.camel.LoggingLevel", "enum": [ "TRACE", "DEBUG", "INFO", "WARN", "ERROR", "OFF" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "TRACE", "description": "The consumer logs a start\/complete log line when it polls. This option allows you to configure the logging level [...] "scheduledExecutorService": { "kind": "parameter", "displayName": "Scheduled Executor Service", "group": "scheduler", "label": "consumer,scheduler", "required": false, "type": "object", "javaType": "java.util.concurrent.ScheduledExecutorService", "deprecated": false, "autowired": false, "secret": false, "description": "Allows for configuring a custom\/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool." }, diff --git a/components/camel-scheduler/src/main/docs/scheduler-component.adoc b/components/camel-scheduler/src/main/docs/scheduler-component.adoc index 96148bd..1a571f0 100644 --- a/components/camel-scheduler/src/main/docs/scheduler-component.adoc +++ b/components/camel-scheduler/src/main/docs/scheduler-component.adoc @@ -51,7 +51,7 @@ The Scheduler component supports 3 options, which are listed below. | Name | Description | Default | Type | *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean | *autowiredEnabled* (advanced) | Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc. | true | boolean -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int |=== // component options: END @@ -90,10 +90,10 @@ with the following path and query parameters: | *backoffErrorThreshold* (scheduler) | The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in. | | int | *backoffIdleThreshold* (scheduler) | The number of subsequent idle polls that should happen before the backoffMultipler should kick-in. | | int | *backoffMultiplier* (scheduler) | To let the scheduled polling consumer backoff if there has been a number of subsequent idles/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening again. When this option is in use then backoffIdleThreshold and/or backoffErrorThreshold must also be configured. | | int -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int | *delay* (scheduler) | Milliseconds before the next poll. | 500 | long | *greedy* (scheduler) | If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages. | false | boolean | *initialDelay* (scheduler) | Milliseconds before the first poll starts. | 1000 | long +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int | *repeatCount* (scheduler) | Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever. | 0 | long | *runLoggingLevel* (scheduler) | The consumer logs a start/complete log line when it polls. This option allows you to configure the logging level for that. There are 6 enums and the value can be one of: TRACE, DEBUG, INFO, WARN, ERROR, OFF | TRACE | LoggingLevel | *scheduledExecutorService* (scheduler) | Allows for configuring a custom/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool. | | ScheduledExecutorService diff --git a/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerComponent.java b/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerComponent.java index 5db0bbe..475044a 100644 --- a/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerComponent.java +++ b/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerComponent.java @@ -33,7 +33,7 @@ public class SchedulerComponent extends DefaultComponent { private final Map<String, AtomicInteger> refCounts = new HashMap<>(); @Metadata(defaultValue = "1", label = "scheduler") - private int concurrentTasks = 1; + private int poolSize = 1; public SchedulerComponent() { } @@ -41,34 +41,34 @@ public class SchedulerComponent extends DefaultComponent { @Override protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception { SchedulerEndpoint answer = new SchedulerEndpoint(uri, this, remaining); - answer.setConcurrentTasks(getConcurrentTasks()); + answer.setPoolSize(getPoolSize()); setProperties(answer, parameters); return answer; } - public int getConcurrentTasks() { - return concurrentTasks; + public int getPoolSize() { + return poolSize; } /** - * Number of threads used by the scheduling thread pool. + * Number of core threads in the thread pool used by the scheduling thread pool. * <p/> * Is by default using a single thread */ - public void setConcurrentTasks(int concurrentTasks) { - this.concurrentTasks = concurrentTasks; + public void setPoolSize(int poolSize) { + this.poolSize = poolSize; } protected ScheduledExecutorService addConsumer(SchedulerConsumer consumer) { String name = consumer.getEndpoint().getName(); - int concurrentTasks = consumer.getEndpoint().getConcurrentTasks(); + int poolSize = consumer.getEndpoint().getPoolSize(); ScheduledExecutorService answer; synchronized (executors) { answer = executors.get(name); if (answer == null) { answer = getCamelContext().getExecutorServiceManager().newScheduledThreadPool(this, "scheduler://" + name, - concurrentTasks); + poolSize); executors.put(name, answer); // store new reference counter refCounts.put(name, new AtomicInteger(1)); diff --git a/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java b/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java index d1528eb..b460e5a 100644 --- a/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java +++ b/components/camel-scheduler/src/main/java/org/apache/camel/component/scheduler/SchedulerEndpoint.java @@ -42,7 +42,7 @@ public class SchedulerEndpoint extends ScheduledPollEndpoint { @Metadata(required = true) private String name; @UriParam(defaultValue = "1", label = "scheduler") - private int concurrentTasks = 1; + private int poolSize = 1; @UriParam(defaultValue = "false", label = "advanced", description = "Sets whether synchronous processing should be strictly used") private boolean synchronous; @@ -80,17 +80,17 @@ public class SchedulerEndpoint extends ScheduledPollEndpoint { this.name = name; } - public int getConcurrentTasks() { - return concurrentTasks; + public int getPoolSize() { + return poolSize; } /** - * Number of threads used by the scheduling thread pool. + * Number of core threads in the thread pool used by the scheduling thread pool. * <p/> * Is by default using a single thread */ - public void setConcurrentTasks(int concurrentTasks) { - this.concurrentTasks = concurrentTasks; + public void setPoolSize(int poolSize) { + this.poolSize = poolSize; } public boolean isSynchronous() { diff --git a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SchedulerComponentBuilderFactory.java b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SchedulerComponentBuilderFactory.java index e59414b..2f5f7ed 100644 --- a/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SchedulerComponentBuilderFactory.java +++ b/core/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/SchedulerComponentBuilderFactory.java @@ -96,19 +96,19 @@ public interface SchedulerComponentBuilderFactory { return this; } /** - * Number of threads used by the scheduling thread pool. Is by default - * using a single thread. + * Number of core threads in the thread pool used by the scheduling + * thread pool. Is by default using a single thread. * * The option is a: <code>int</code> type. * * Default: 1 * Group: scheduler * - * @param concurrentTasks the value to set + * @param poolSize the value to set * @return the dsl builder */ - default SchedulerComponentBuilder concurrentTasks(int concurrentTasks) { - doSetProperty("concurrentTasks", concurrentTasks); + default SchedulerComponentBuilder poolSize(int poolSize) { + doSetProperty("poolSize", poolSize); return this; } } @@ -130,7 +130,7 @@ public interface SchedulerComponentBuilderFactory { switch (name) { case "bridgeErrorHandler": ((SchedulerComponent) component).setBridgeErrorHandler((boolean) value); return true; case "autowiredEnabled": ((SchedulerComponent) component).setAutowiredEnabled((boolean) value); return true; - case "concurrentTasks": ((SchedulerComponent) component).setConcurrentTasks((int) value); return true; + case "poolSize": ((SchedulerComponent) component).setPoolSize((int) value); return true; default: return false; } } diff --git a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerBlockingTest.java b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerBlockingTest.java new file mode 100644 index 0000000..674e316 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerBlockingTest.java @@ -0,0 +1,44 @@ +package org.apache.camel.component.scheduler; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; + +@Disabled("Manual test") +public class SchedulerBlockingTest extends ContextTestSupport { + + @Test + public void testScheduler() throws Exception { + Thread.sleep(60000); + } + + @Override + protected RouteBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + @Override + public void configure() throws Exception { + SchedulerComponent comp = getContext().getComponent("scheduler", SchedulerComponent.class); + comp.setPoolSize(4); + + from("scheduler://trigger?delay=2000&repeatCount=3").routeId("scheduler") + .threads(10) + .log("1") + .inOut("seda:route1") + .log("1.1"); + + from("seda:route1?concurrentConsumers=2").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"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerNoPolledMessagesTest.java b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerNoPolledMessagesTest.java index 2833df3..ecc4f07 100644 --- a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerNoPolledMessagesTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/SchedulerNoPolledMessagesTest.java @@ -42,7 +42,7 @@ public class SchedulerNoPolledMessagesTest extends ContextTestSupport { protected RouteBuilder createRouteBuilder() throws Exception { return new RouteBuilder() { public void configure() { - from("scheduler://foo?delay=100&backoffMultiplier=10&backoffIdleThreshold=2&scheduler.concurrentTasks=2") + from("scheduler://foo?delay=100&backoffMultiplier=10&backoffIdleThreshold=2&poolSize=2") .log("Fired scheduler").process(new Processor() { @Override public void process(Exchange exchange) throws Exception { diff --git a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksOneRouteTest.java b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksOneRouteTest.java index 305512d..6918876 100644 --- a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksOneRouteTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksOneRouteTest.java @@ -39,13 +39,11 @@ public class TwoSchedulerConcurrentTasksOneRouteTest extends ContextTestSupport protected RouteBuilder createRouteBuilder() throws Exception { return new RouteBuilder() { public void configure() { - // number of concurrent task a thread pool should have + // number of core threads a thread pool should have SchedulerComponent comp = context.getComponent("scheduler", SchedulerComponent.class); - comp.setConcurrentTasks(2); + comp.setPoolSize(2); - // let this route scheduler use all 2 concurrent tasks at the - // same time - from("scheduler://foo?delay=250&scheduler.concurrentTasks=2").process(new Processor() { + from("scheduler://foo?delay=250").process(new Processor() { @Override public void process(Exchange exchange) throws Exception { if (sleep.compareAndSet(true, false)) { diff --git a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksTest.java b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksTest.java index 2443208..66e3eed 100644 --- a/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksTest.java +++ b/core/camel-core/src/test/java/org/apache/camel/component/scheduler/TwoSchedulerConcurrentTasksTest.java @@ -35,7 +35,7 @@ public class TwoSchedulerConcurrentTasksTest extends ContextTestSupport { return new RouteBuilder() { public void configure() { SchedulerComponent comp = context.getComponent("scheduler", SchedulerComponent.class); - comp.setConcurrentTasks(2); + comp.setPoolSize(2); from("scheduler://foo?delay=100").to("log:a").to("mock:a"); diff --git a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SchedulerEndpointBuilderFactory.java b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SchedulerEndpointBuilderFactory.java index 5bf11c9..f3edaf5 100644 --- a/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SchedulerEndpointBuilderFactory.java +++ b/core/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/SchedulerEndpointBuilderFactory.java @@ -229,38 +229,6 @@ public interface SchedulerEndpointBuilderFactory { return this; } /** - * Number of threads used by the scheduling thread pool. Is by default - * using a single thread. - * - * The option is a: <code>int</code> type. - * - * Default: 1 - * Group: scheduler - * - * @param concurrentTasks the value to set - * @return the dsl builder - */ - default SchedulerEndpointBuilder concurrentTasks(int concurrentTasks) { - doSetProperty("concurrentTasks", concurrentTasks); - return this; - } - /** - * Number of threads used by the scheduling thread pool. Is by default - * using a single thread. - * - * The option will be converted to a <code>int</code> type. - * - * Default: 1 - * Group: scheduler - * - * @param concurrentTasks the value to set - * @return the dsl builder - */ - default SchedulerEndpointBuilder concurrentTasks(String concurrentTasks) { - doSetProperty("concurrentTasks", concurrentTasks); - return this; - } - /** * Milliseconds before the next poll. * * The option is a: <code>long</code> type. @@ -354,6 +322,38 @@ public interface SchedulerEndpointBuilderFactory { return this; } /** + * Number of core threads in the thread pool used by the scheduling + * thread pool. Is by default using a single thread. + * + * The option is a: <code>int</code> type. + * + * Default: 1 + * Group: scheduler + * + * @param poolSize the value to set + * @return the dsl builder + */ + default SchedulerEndpointBuilder poolSize(int poolSize) { + doSetProperty("poolSize", poolSize); + return this; + } + /** + * Number of core threads in the thread pool used by the scheduling + * thread pool. Is by default using a single thread. + * + * The option will be converted to a <code>int</code> type. + * + * Default: 1 + * Group: scheduler + * + * @param poolSize the value to set + * @return the dsl builder + */ + default SchedulerEndpointBuilder poolSize(String poolSize) { + doSetProperty("poolSize", poolSize); + return this; + } + /** * Specifies a maximum limit of number of fires. So if you set it to 1, * the scheduler will only fire once. If you set it to 5, it will only * fire five times. A value of zero or negative means fire forever. 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 3a7e253..af6aa81 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 @@ -47,7 +47,7 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem private boolean shutdownExecutor; private volatile List<ScheduledFuture<?>> futures = new ArrayList<>(); private Runnable task; - private int concurrentTasks = 1; + private int poolSize = 1; private long initialDelay = -1; private long delay = -1; @@ -111,12 +111,12 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem this.scheduledExecutorService = scheduledExecutorService; } - public int getConcurrentTasks() { - return concurrentTasks; + public int getPoolSize() { + return poolSize; } - public void setConcurrentTasks(int concurrentTasks) { - this.concurrentTasks = concurrentTasks; + public void setPoolSize(int poolSize) { + this.poolSize = poolSize; } @Override @@ -170,20 +170,16 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem currentInitialDelay, currentDelay, getTimeUnit().name().toLowerCase(Locale.ENGLISH), consumer.getEndpoint()); } - for (int i = 0; i < concurrentTasks; i++) { - futures.add(scheduledExecutorService.scheduleWithFixedDelay(task, currentInitialDelay, currentDelay, - getTimeUnit())); - } + 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()); } - for (int i = 0; i < concurrentTasks; i++) { - futures.add(scheduledExecutorService.scheduleAtFixedRate(task, currentInitialDelay, currentDelay, - getTimeUnit())); - } + futures.add(scheduledExecutorService.scheduleAtFixedRate(task, currentInitialDelay, currentDelay, + getTimeUnit())); } } } @@ -203,7 +199,7 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem if (scheduledExecutorService == null) { // we only need one thread in the pool to schedule this task this.scheduledExecutorService = getCamelContext().getExecutorServiceManager() - .newScheduledThreadPool(consumer, consumer.getEndpoint().getEndpointUri(), concurrentTasks); + .newScheduledThreadPool(consumer, consumer.getEndpoint().getEndpointUri(), poolSize); // and we should shutdown the thread pool when no longer needed this.shutdownExecutor = true; } diff --git a/docs/components/modules/ROOT/pages/scheduler-component.adoc b/docs/components/modules/ROOT/pages/scheduler-component.adoc index abab3ed..8f13478 100644 --- a/docs/components/modules/ROOT/pages/scheduler-component.adoc +++ b/docs/components/modules/ROOT/pages/scheduler-component.adoc @@ -53,7 +53,7 @@ The Scheduler component supports 3 options, which are listed below. | Name | Description | Default | Type | *bridgeErrorHandler* (consumer) | Allows for bridging the consumer to the Camel routing Error Handler, which mean any exceptions occurred while the consumer is trying to pickup incoming messages, or the likes, will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions, that will be logged at WARN or ERROR level and ignored. | false | boolean | *autowiredEnabled* (advanced) | Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which then gets configured on the component. This can be used for automatic configuring JDBC data sources, JMS connection factories, AWS Clients, etc. | true | boolean -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int |=== // component options: END @@ -92,10 +92,10 @@ with the following path and query parameters: | *backoffErrorThreshold* (scheduler) | The number of subsequent error polls (failed due some error) that should happen before the backoffMultipler should kick-in. | | int | *backoffIdleThreshold* (scheduler) | The number of subsequent idle polls that should happen before the backoffMultipler should kick-in. | | int | *backoffMultiplier* (scheduler) | To let the scheduled polling consumer backoff if there has been a number of subsequent idles/errors in a row. The multiplier is then the number of polls that will be skipped before the next actual attempt is happening again. When this option is in use then backoffIdleThreshold and/or backoffErrorThreshold must also be configured. | | int -| *concurrentTasks* (scheduler) | Number of threads used by the scheduling thread pool. Is by default using a single thread | 1 | int | *delay* (scheduler) | Milliseconds before the next poll. | 500 | long | *greedy* (scheduler) | If greedy is enabled, then the ScheduledPollConsumer will run immediately again, if the previous run polled 1 or more messages. | false | boolean | *initialDelay* (scheduler) | Milliseconds before the first poll starts. | 1000 | long +| *poolSize* (scheduler) | Number of core threads in the thread pool used by the scheduling thread pool. Is by default using a single thread | 1 | int | *repeatCount* (scheduler) | Specifies a maximum limit of number of fires. So if you set it to 1, the scheduler will only fire once. If you set it to 5, it will only fire five times. A value of zero or negative means fire forever. | 0 | long | *runLoggingLevel* (scheduler) | The consumer logs a start/complete log line when it polls. This option allows you to configure the logging level for that. There are 6 enums and the value can be one of: TRACE, DEBUG, INFO, WARN, ERROR, OFF | TRACE | LoggingLevel | *scheduledExecutorService* (scheduler) | Allows for configuring a custom/shared thread pool to use for the consumer. By default each consumer has its own single threaded thread pool. | | ScheduledExecutorService diff --git a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_10.adoc b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_10.adoc new file mode 100644 index 0000000..3dedb00 --- /dev/null +++ b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide-3_10.adoc @@ -0,0 +1,19 @@ += Apache Camel 3.x Upgrade Guide + +This document is for helping you upgrade your Apache Camel application +from Camel 3.x to 3.y. For example if you are upgrading Camel 3.0 to 3.2, then you should follow the guides +from both 3.0 to 3.1 and 3.1 to 3.2. + +== Upgrading Camel 3.9 to 3.10 + +=== camel-scheduler + +The option `concurrentTasks` has been renamed to `poolSize` to better reflect its purpose. +The scheduler has also been fixed to only schedule triggering by one, and not as mistakenly by causing +concurrent triggering which causes routing mistakes. + +The option configures the pool size of the scheduled thread pool used by the scheduler. + +If the scheduler thread is being blocked by anywhere in the downstream routing, and you want the scheduler +to schedule with a fixed interval, you can use the Threads EIP as queue for pending tasks. + diff --git a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide.adoc b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide.adoc index 2b46f95..1639c9d 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-3x-upgrade-guide.adoc @@ -18,3 +18,4 @@ You can find upgrade guide for each release in the following pages: - xref:camel-3x-upgrade-guide-3_7.adoc[Upgrade guide from 3.6 to 3.7] - xref:camel-3x-upgrade-guide-3_8.adoc[Upgrade guide from 3.7 to 3.8] - xref:camel-3x-upgrade-guide-3_9.adoc[Upgrade guide from 3.8 to 3.9] +- xref:camel-3x-upgrade-guide-3_10.adoc[Upgrade guide from 3.9 to 3.10]
