This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 0d861d06b5b CAMEL-22206: Do not use BackOff but BackgroundTask for
reconnector (#18494)
0d861d06b5b is described below
commit 0d861d06b5b89525bad0287a116e6a941bd34981
Author: Claus Ibsen <[email protected]>
AuthorDate: Sat Jun 28 11:29:54 2025 +0200
CAMEL-22206: Do not use BackOff but BackgroundTask for reconnector (#18494)
* CAMEL-22206: camel-pgevent should not use BackOff but BackgroundTask for
reconnector
* CAMEL-22206: camel-master should not use BackOff but BackgroundTask for
reconnector
* CAMEL-22206: camel-sjms should not use BackOff but BackgroundTask for
reconnector
---
.../apache/camel/catalog/components/master.json | 2 +-
.../master/MasterComponentConfigurer.java | 4 +-
.../org/apache/camel/component/master/master.json | 2 +-
.../camel/component/master/MasterComponent.java | 31 +--------
.../camel/component/master/MasterConsumer.java | 52 +++++++++------
.../camel/component/master/MasterEndpoint.java | 11 +---
.../component/master/MasterComponentTest.java | 8 ++-
.../camel/component/pgevent/PgEventConsumer.java | 51 +++++++++------
.../consumer/SimpleMessageListenerContainer.java | 76 ++++++++++++----------
.../apache/camel/support/task/BackgroundTask.java | 21 ++++++
.../apache/camel/support/task/BlockingTask.java | 7 ++
.../apache/camel/support/task/ForegroundTask.java | 12 +++-
.../dsl/MasterComponentBuilderFactory.java | 6 +-
13 files changed, 164 insertions(+), 119 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/master.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/master.json
index 840a8e99738..0d27e7fa776 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/master.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/master.json
@@ -27,7 +27,7 @@
"bridgeErrorHandler": { "index": 0, "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 (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"autowiredEnabled": { "index": 1, "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 t [...]
"backOffDelay": { "index": 2, "kind": "property", "displayName": "Back Off
Delay", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "description": "When the master becomes leader then backoff is
in use to repeat starting the consumer until the consumer is successfully
started or max attempts reached. This option is the delay in millis between
start attempts." },
- "backOffMaxAttempts": { "index": 3, "kind": "property", "displayName":
"Back Off Max Attempts", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "description": "When the master becomes leader then
backoff is in use to repeat starting the consumer until the consumer is
successfully started or max attempts reached. This option is the maximum number
of attempts to try." },
+ "backOffMaxAttempts": { "index": 3, "kind": "property", "displayName":
"Back Off Max Attempts", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "description": "When the master becomes leader then
backoff is in use to repeat starting the consumer until the consumer is
successfully started or max attempts reached. This option is the maximum number
of attempts to try." },
"service": { "index": 4, "kind": "property", "displayName": "Service",
"group": "advanced", "label": "advanced", "required": false, "type": "object",
"javaType": "org.apache.camel.cluster.CamelClusterService", "deprecated":
false, "autowired": false, "secret": false, "description": "Inject the service
to use." },
"serviceSelector": { "index": 5, "kind": "property", "displayName":
"Service Selector", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.cluster.CamelClusterService.Selector", "deprecated": false,
"autowired": false, "secret": false, "description": "Inject the service
selector used to lookup the CamelClusterService to use." }
},
diff --git
a/components/camel-master/src/generated/java/org/apache/camel/component/master/MasterComponentConfigurer.java
b/components/camel-master/src/generated/java/org/apache/camel/component/master/MasterComponentConfigurer.java
index 81a7ecfe5e7..2d0413d4bcd 100644
---
a/components/camel-master/src/generated/java/org/apache/camel/component/master/MasterComponentConfigurer.java
+++
b/components/camel-master/src/generated/java/org/apache/camel/component/master/MasterComponentConfigurer.java
@@ -28,7 +28,7 @@ public class MasterComponentConfigurer extends
PropertyConfigurerSupport impleme
case "backoffdelay":
case "backOffDelay": target.setBackOffDelay(property(camelContext,
long.class, value)); return true;
case "backoffmaxattempts":
- case "backOffMaxAttempts":
target.setBackOffMaxAttempts(property(camelContext, long.class, value)); return
true;
+ case "backOffMaxAttempts":
target.setBackOffMaxAttempts(property(camelContext, int.class, value)); return
true;
case "bridgeerrorhandler":
case "bridgeErrorHandler":
target.setBridgeErrorHandler(property(camelContext, boolean.class, value));
return true;
case "service": target.setService(property(camelContext,
org.apache.camel.cluster.CamelClusterService.class, value)); return true;
@@ -46,7 +46,7 @@ public class MasterComponentConfigurer extends
PropertyConfigurerSupport impleme
case "backoffdelay":
case "backOffDelay": return long.class;
case "backoffmaxattempts":
- case "backOffMaxAttempts": return long.class;
+ case "backOffMaxAttempts": return int.class;
case "bridgeerrorhandler":
case "bridgeErrorHandler": return boolean.class;
case "service": return
org.apache.camel.cluster.CamelClusterService.class;
diff --git
a/components/camel-master/src/generated/resources/META-INF/org/apache/camel/component/master/master.json
b/components/camel-master/src/generated/resources/META-INF/org/apache/camel/component/master/master.json
index 840a8e99738..0d27e7fa776 100644
---
a/components/camel-master/src/generated/resources/META-INF/org/apache/camel/component/master/master.json
+++
b/components/camel-master/src/generated/resources/META-INF/org/apache/camel/component/master/master.json
@@ -27,7 +27,7 @@
"bridgeErrorHandler": { "index": 0, "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 (if possible) occurred while the Camel consumer is trying
to pickup incoming messages, or the like [...]
"autowiredEnabled": { "index": 1, "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 t [...]
"backOffDelay": { "index": 2, "kind": "property", "displayName": "Back Off
Delay", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "long", "deprecated": false, "autowired": false,
"secret": false, "description": "When the master becomes leader then backoff is
in use to repeat starting the consumer until the consumer is successfully
started or max attempts reached. This option is the delay in millis between
start attempts." },
- "backOffMaxAttempts": { "index": 3, "kind": "property", "displayName":
"Back Off Max Attempts", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "long", "deprecated": false, "autowired":
false, "secret": false, "description": "When the master becomes leader then
backoff is in use to repeat starting the consumer until the consumer is
successfully started or max attempts reached. This option is the maximum number
of attempts to try." },
+ "backOffMaxAttempts": { "index": 3, "kind": "property", "displayName":
"Back Off Max Attempts", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "description": "When the master becomes leader then
backoff is in use to repeat starting the consumer until the consumer is
successfully started or max attempts reached. This option is the maximum number
of attempts to try." },
"service": { "index": 4, "kind": "property", "displayName": "Service",
"group": "advanced", "label": "advanced", "required": false, "type": "object",
"javaType": "org.apache.camel.cluster.CamelClusterService", "deprecated":
false, "autowired": false, "secret": false, "description": "Inject the service
to use." },
"serviceSelector": { "index": 5, "kind": "property", "displayName":
"Service Selector", "group": "advanced", "label": "advanced", "required":
false, "type": "object", "javaType":
"org.apache.camel.cluster.CamelClusterService.Selector", "deprecated": false,
"autowired": false, "secret": false, "description": "Inject the service
selector used to lookup the CamelClusterService to use." }
},
diff --git
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterComponent.java
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterComponent.java
index b4d8bc4e59f..95ddbd92434 100644
---
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterComponent.java
+++
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterComponent.java
@@ -17,7 +17,6 @@
package org.apache.camel.component.master;
import java.util.Map;
-import java.util.concurrent.ScheduledExecutorService;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
@@ -46,9 +45,7 @@ public class MasterComponent extends DefaultComponent {
@Metadata(label = "advanced")
private long backOffDelay = 5000;
@Metadata(label = "advanced")
- private long backOffMaxAttempts = 10;
-
- private ScheduledExecutorService backOffThreadPool;
+ private int backOffMaxAttempts = 10;
public MasterComponent() {
this(null);
@@ -111,10 +108,6 @@ public class MasterComponent extends DefaultComponent {
this.serviceSelector = serviceSelector;
}
- public ScheduledExecutorService getBackOffThreadPool() {
- return backOffThreadPool;
- }
-
public long getBackOffDelay() {
return backOffDelay;
}
@@ -129,7 +122,7 @@ public class MasterComponent extends DefaultComponent {
this.backOffDelay = backOffDelay;
}
- public long getBackOffMaxAttempts() {
+ public int getBackOffMaxAttempts() {
return backOffMaxAttempts;
}
@@ -139,7 +132,7 @@ public class MasterComponent extends DefaultComponent {
*
* This option is the maximum number of attempts to try.
*/
- public void setBackOffMaxAttempts(long backOffMaxAttempts) {
+ public void setBackOffMaxAttempts(int backOffMaxAttempts) {
this.backOffMaxAttempts = backOffMaxAttempts;
}
@@ -153,22 +146,4 @@ public class MasterComponent extends DefaultComponent {
() -> new IllegalStateException("No cluster service
found"));
}
}
-
- @Override
- protected void doStart() throws Exception {
- if (backOffThreadPool == null) {
- backOffThreadPool
- =
getCamelContext().getExecutorServiceManager().newDefaultScheduledThreadPool(this,
"MasterLeaderTask");
- }
- }
-
- @Override
- protected void doStop() throws Exception {
- super.doStop();
-
- if (backOffThreadPool != null) {
-
getCamelContext().getExecutorServiceManager().shutdown(backOffThreadPool);
- backOffThreadPool = null;
- }
- }
}
diff --git
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
index 652c82e2865..603fa9d5073 100644
---
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
+++
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterConsumer.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.master;
+import java.time.Duration;
+import java.util.concurrent.ScheduledExecutorService;
+
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Processor;
@@ -31,11 +34,11 @@ import org.apache.camel.resume.ResumeAdapter;
import org.apache.camel.resume.ResumeAware;
import org.apache.camel.resume.ResumeStrategy;
import org.apache.camel.support.DefaultConsumer;
-import org.apache.camel.support.PluginHelper;
import org.apache.camel.support.resume.AdapterHelper;
import org.apache.camel.support.service.ServiceHelper;
-import org.apache.camel.util.backoff.BackOff;
-import org.apache.camel.util.backoff.BackOffTimer;
+import org.apache.camel.support.task.BackgroundTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,11 +57,10 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
private volatile Consumer delegatedConsumer;
private volatile CamelClusterView view;
private ResumeStrategy resumeStrategy;
- private BackOffTimer timer;
+ private ScheduledExecutorService leaderPool;
public MasterConsumer(MasterEndpoint masterEndpoint, Processor processor,
CamelClusterService clusterService) {
super(masterEndpoint, processor);
-
this.clusterService = clusterService;
this.masterEndpoint = masterEndpoint;
this.delegatedEndpoint = masterEndpoint.getEndpoint();
@@ -79,16 +81,16 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
@Override
protected void doInit() throws Exception {
super.doInit();
- this.timer =
PluginHelper.getBackOffTimerFactory(masterEndpoint.getCamelContext().getCamelContextExtension())
- .newBackOffTimer("MasterConsumer",
masterEndpoint.getComponent().getBackOffThreadPool());
+
+ // used for re-connecting to the database
+ leaderPool =
getEndpoint().getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this, "Leadership");
}
@Override
protected void doStart() throws Exception {
super.doStart();
- ServiceHelper.startService(timer);
-
LOG.debug("Using ClusterService instance {} (id={}, type={})",
clusterService, clusterService.getId(),
clusterService.getClass().getName());
@@ -106,7 +108,8 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
view = null;
}
- ServiceHelper.stopAndShutdownServices(delegatedConsumer,
delegatedEndpoint, timer);
+
getEndpoint().getCamelContext().getExecutorServiceManager().shutdown(leaderPool);
+ ServiceHelper.stopAndShutdownServices(delegatedConsumer,
delegatedEndpoint);
delegatedConsumer = null;
}
@@ -135,6 +138,18 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
// Helpers
// **************************************
+ private BackgroundTask createTask() {
+ return Tasks.backgroundTask()
+ .withScheduledExecutor(leaderPool)
+ .withBudget(Budgets.iterationTimeBudget()
+
.withInterval(Duration.ofMillis(masterEndpoint.getComponent().getBackOffDelay()))
+ .withInitialDelay(Duration.ofSeconds(1))
+
.withMaxIterations(masterEndpoint.getComponent().getBackOffMaxAttempts())
+ .build())
+ .withName("Leadership")
+ .build();
+ }
+
private void onLeadershipTaken() throws Exception {
lock.lock();
try {
@@ -146,13 +161,12 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
return;
}
- // start consumer using background task up till X attempts
- long delay = masterEndpoint.getComponent().getBackOffDelay();
- long max = masterEndpoint.getComponent().getBackOffMaxAttempts();
-
-
timer.schedule(BackOff.builder().delay(delay).maxAttempts(max).build(), task ->
{
- LOG.info("Leadership taken. Attempt #{} to start consumer:
{}", task.getCurrentAttempts(),
- delegatedEndpoint);
+ final BackgroundTask leaderTask = createTask();
+ leaderTask.run(() -> {
+ if (!isRunAllowed()) {
+ return false;
+ }
+ LOG.info("Leadership taken. Attempt #{} to start consumer:
{}", leaderTask.iteration(), delegatedEndpoint);
Exception cause = null;
try {
@@ -179,13 +193,13 @@ public class MasterConsumer extends DefaultConsumer
implements ResumeAware<Resum
}
if (cause != null) {
- String message = "Leadership taken. Attempt #" +
task.getCurrentAttempts()
+ String message = "Leadership taken. Attempt #" +
leaderTask.iteration()
+ " failed to start consumer due to: " +
cause.getMessage();
getExceptionHandler().handleException(message, cause);
return true; // retry
}
- LOG.info("Leadership taken. Attempt #" +
task.getCurrentAttempts() + " success. Consumer started: {}",
+ LOG.info("Leadership taken. Attempt #{} success. Consumer
started: {}", leaderTask.iteration(),
delegatedEndpoint);
return false; // no more attempts
});
diff --git
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterEndpoint.java
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterEndpoint.java
index ff52293fe4b..64e22c5ae75 100644
---
a/components/camel-master/src/main/java/org/apache/camel/component/master/MasterEndpoint.java
+++
b/components/camel-master/src/main/java/org/apache/camel/component/master/MasterEndpoint.java
@@ -34,14 +34,10 @@ import org.apache.camel.support.DefaultEndpoint;
* Have only a single consumer in a cluster consuming from a given endpoint;
with automatic failover if the JVM dies.
*/
@ManagedResource(description = "Managed Master Endpoint")
-@UriEndpoint(firstVersion = "2.20.0",
- scheme = "master",
- syntax = "master:namespace:delegateUri",
- consumerOnly = true,
- title = "Master",
- lenientProperties = true,
- category = { Category.CLUSTERING })
+@UriEndpoint(firstVersion = "2.20.0", scheme = "master", syntax =
"master:namespace:delegateUri", title = "Master",
+ consumerOnly = true, lenientProperties = true, category = {
Category.CLUSTERING })
public class MasterEndpoint extends DefaultEndpoint implements
DelegateEndpoint {
+
private final Endpoint delegateEndpoint;
private final CamelClusterService clusterService;
@@ -56,7 +52,6 @@ public class MasterEndpoint extends DefaultEndpoint
implements DelegateEndpoint
public MasterEndpoint(String uri, MasterComponent component,
CamelClusterService clusterService, String namespace,
String delegateUri) {
super(uri, component);
-
this.clusterService = clusterService;
this.namespace = namespace;
this.delegateUri = delegateUri;
diff --git
a/components/camel-master/src/test/java/org/apache/camel/component/master/MasterComponentTest.java
b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterComponentTest.java
index 5fa7240cf08..6e069c1e644 100644
---
a/components/camel-master/src/test/java/org/apache/camel/component/master/MasterComponentTest.java
+++
b/components/camel-master/src/test/java/org/apache/camel/component/master/MasterComponentTest.java
@@ -90,11 +90,15 @@ public class MasterComponentTest {
// Start the context after some random time so the startup order
// changes for each test.
Awaitility.await().pollDelay(ThreadLocalRandom.current().nextInt(500),
TimeUnit.MILLISECONDS)
- .untilAsserted(() ->
Assertions.assertDoesNotThrow(context::start));
+ .untilAsserted(() -> Assertions.assertDoesNotThrow(() -> {
+ LOGGER.info("Starting node {}", id);
+ context.start();
+ }));
+ LOGGER.info("Waiting for {} events on node {}", events, id);
contextLatch.await();
- LOGGER.debug("Shutting down node {}", id);
+ LOGGER.info("Shutting down node {}", id);
RESULTS.add(id);
context.stop();
diff --git
a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
index e5d52ef9db1..666b61f4a1c 100644
---
a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
+++
b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.pgevent;
import java.sql.PreparedStatement;
+import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
@@ -26,10 +27,9 @@ import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.support.DefaultConsumer;
-import org.apache.camel.support.PluginHelper;
-import org.apache.camel.support.service.ServiceHelper;
-import org.apache.camel.util.backoff.BackOff;
-import org.apache.camel.util.backoff.BackOffTimer;
+import org.apache.camel.support.task.BackgroundTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,9 +44,9 @@ public class PgEventConsumer extends DefaultConsumer {
private final PgEventEndpoint endpoint;
private PGConnection dbConnection;
private ScheduledExecutorService reconnectPool;
+ private BackgroundTask reconnectTask;
private ExecutorService workerPool;
private boolean shutdownWorkerPool;
- private BackOffTimer timer;
public PgEventConsumer(PgEventEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -67,14 +67,20 @@ public class PgEventConsumer extends DefaultConsumer {
}
// used for re-connecting to the database
reconnectPool =
getEndpoint().getCamelContext().getExecutorServiceManager()
- .newSingleThreadScheduledExecutor(this, "Reconnector");
- timer =
PluginHelper.getBackOffTimerFactory(endpoint.getCamelContext().getCamelContextExtension())
- .newBackOffTimer("PgEventConsumer", reconnectPool);
+ .newSingleThreadScheduledExecutor(this, "PgEventReconnect");
+ reconnectTask = Tasks.backgroundTask()
+ .withScheduledExecutor(reconnectPool)
+ .withBudget(Budgets.iterationTimeBudget()
+
.withInterval(Duration.ofMillis(endpoint.getReconnectDelay()))
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withUnlimitedDuration()
+ .build())
+ .withName("PgEventReconnect")
+ .build();
}
@Override
protected void doStart() throws Exception {
- ServiceHelper.startService(timer);
listener.initConnection();
super.doStart();
}
@@ -89,23 +95,26 @@ public class PgEventConsumer extends DefaultConsumer {
endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(workerPool,
10000);
workerPool = null;
}
- ServiceHelper.stopService(timer);
}
public class PgEventListener implements PGNotificationListener {
public void reconnect() {
- BackOff bo =
BackOff.builder().delay(endpoint.getReconnectDelay()).build();
- timer.schedule(bo, t -> {
- LOG.debug("Connecting attempt #{}", t.getCurrentAttempts());
- try {
- initConnection();
- } catch (Exception e) {
- return true;
- }
- LOG.debug("Connecting successful");
- return false;
- });
+ // only submit the task if not already running
+ if (!reconnectTask.isRunning()) {
+ reconnectTask.run(() -> {
+ if (isRunAllowed()) {
+ LOG.debug("Connecting attempt #{}",
reconnectTask.iteration());
+ try {
+ initConnection();
+ } catch (Exception e) {
+ return true;
+ }
+ LOG.debug("Connecting successful");
+ }
+ return false;
+ });
+ }
}
public void initConnection() throws Exception {
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
index bcf9c56fc91..23723daf9a4 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/SimpleMessageListenerContainer.java
@@ -16,8 +16,10 @@
*/
package org.apache.camel.component.sjms.consumer;
+import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
+import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -35,11 +37,10 @@ import org.apache.camel.CamelContext;
import org.apache.camel.component.sjms.SessionMessageListener;
import org.apache.camel.component.sjms.SjmsEndpoint;
import org.apache.camel.component.sjms.jms.DestinationCreationStrategy;
-import org.apache.camel.support.PluginHelper;
-import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.support.service.ServiceSupport;
-import org.apache.camel.util.backoff.BackOff;
-import org.apache.camel.util.backoff.BackOffTimer;
+import org.apache.camel.support.task.BackgroundTask;
+import org.apache.camel.support.task.Tasks;
+import org.apache.camel.support.task.budget.Budgets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,9 +65,9 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
private final Lock consumerLock = new ReentrantLock();
private Set<MessageConsumer> consumers;
private Set<Session> sessions;
- private BackOffTimer.Task recoverTask;
- private BackOffTimer timer;
- private ScheduledExecutorService scheduler;
+ private ScheduledExecutorService recoverPool;
+ private BackgroundTask recoverTask;
+ private Future<?> recoverFuture;
public SimpleMessageListenerContainer(SjmsEndpoint endpoint) {
this.endpoint = endpoint;
@@ -196,17 +197,17 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
scheduleConnectionRecovery();
}
- protected boolean recoverConnection(BackOffTimer.Task task) throws
Exception {
- LOG.debug("Recovering from JMS Connection exception (attempt: {})",
task.getCurrentAttempts());
+ protected boolean recoverConnection(BackgroundTask task) {
+ LOG.debug("Recovering from JMS Connection exception (attempt: {})",
task.iteration());
try {
refreshConnection();
initConsumers();
- LOG.debug("Successfully recovered JMS Connection (attempt: {})",
task.getCurrentAttempts());
+ LOG.debug("Successfully recovered JMS Connection (attempt: {})",
task.iteration());
// success so do not try again
return false;
} catch (Exception e) {
if (LOG.isDebugEnabled()) {
- LOG.debug("Failed to recover JMS Connection. Will try again in
{} millis", task.getCurrentDelay(), e);
+ LOG.debug("Failed to recover JMS Connection. Will try again in
{} millis", endpoint.getRecoveryInterval(), e);
}
// try again
return true;
@@ -214,24 +215,33 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
}
protected void scheduleConnectionRecovery() {
- if (scheduler == null) {
- this.scheduler =
endpoint.getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
- "SimpleMessageListenerContainer");
- }
-
- // we need to recover using a background task
- if (recoverTask == null || recoverTask.getStatus() !=
BackOffTimer.Task.Status.Active) {
- BackOff backOff =
BackOff.builder().delay(endpoint.getRecoveryInterval()).build();
- if (timer == null) {
- timer =
PluginHelper.getBackOffTimerFactory(endpoint.getCamelContext().getCamelContextExtension())
- .newBackOffTimer("SjmsConnectionRecovery",
- scheduler);
- ServiceHelper.startService(timer);
+ connectionLock.lock();
+ try {
+ if (recoverPool == null) {
+ recoverPool =
endpoint.getCamelContext().getExecutorServiceManager().newSingleThreadScheduledExecutor(this,
+ "SjmsConnectionRecovery");
+ }
+ if (recoverTask == null) {
+ recoverTask = createTask();
+ recoverFuture = recoverTask.schedule(() ->
recoverConnection(recoverTask));
}
- recoverTask = timer.schedule(backOff, this::recoverConnection);
+ } finally {
+ connectionLock.unlock();
}
}
+ private BackgroundTask createTask() {
+ return Tasks.backgroundTask()
+ .withScheduledExecutor(recoverPool)
+ .withBudget(Budgets.iterationTimeBudget()
+
.withInterval(Duration.ofMillis(endpoint.getRecoveryInterval()))
+ .withInitialDelay(Duration.ofSeconds(1))
+ .withUnlimitedDuration()
+ .build())
+ .withName("SjmsConnectionRecovery")
+ .build();
+ }
+
@Override
protected void doStart() throws Exception {
createConnection();
@@ -242,17 +252,17 @@ public class SimpleMessageListenerContainer extends
ServiceSupport
@Override
protected void doStop() throws Exception {
- if (recoverTask != null) {
- recoverTask.cancel();
- }
stopConnection();
stopConsumers();
- if (scheduler != null) {
-
endpoint.getCamelContext().getExecutorServiceManager().shutdown(scheduler);
- scheduler = null;
+ if (recoverPool != null) {
+
endpoint.getCamelContext().getExecutorServiceManager().shutdown(recoverPool);
+ recoverPool = null;
+ }
+ if (recoverFuture != null && recoverTask != null &&
recoverTask.isRunning()) {
+ recoverFuture.cancel(true);
+ recoverTask = null;
+ recoverFuture = null;
}
- ServiceHelper.stopService(timer);
- timer = null;
}
@Override
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
index 3df28497529..ff829d31ec1 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/BackgroundTask.java
@@ -77,6 +77,7 @@ public class BackgroundTask implements BlockingTask {
private final String name;
private final CountDownLatch latch = new CountDownLatch(1);
private Duration elapsed = Duration.ZERO;
+ private final AtomicBoolean running = new AtomicBoolean();
private final AtomicBoolean completed = new AtomicBoolean();
BackgroundTask(TimeBudget budget, ScheduledExecutorService service, String
name) {
@@ -105,8 +106,22 @@ public class BackgroundTask implements BlockingTask {
}
}
+ /**
+ * Schedules the task to be run
+ *
+ * @param supplier the task as a boolean supplier. The result is used to
check if the task has completed or not.
+ * The supplier must return true if the execution has
completed or false otherwise.
+ * @return a future for the task
+ */
+ public Future<?> schedule(BooleanSupplier supplier) {
+ running.set(true);
+ return service.scheduleAtFixedRate(() -> runTaskWrapper(supplier),
budget.initialDelay(),
+ budget.interval(), TimeUnit.MILLISECONDS);
+ }
+
@Override
public boolean run(BooleanSupplier supplier) {
+ running.set(true);
Future<?> task = service.scheduleAtFixedRate(() ->
runTaskWrapper(supplier), budget.initialDelay(),
budget.interval(), TimeUnit.MILLISECONDS);
waitForTaskCompletion(task);
@@ -133,9 +148,15 @@ public class BackgroundTask implements BlockingTask {
Thread.currentThread().interrupt();
} finally {
elapsed = budget.elapsed();
+ running.set(false);
}
}
+ @Override
+ public boolean isRunning() {
+ return running.get();
+ }
+
@Override
public Duration elapsed() {
return elapsed;
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
index 54b61116c93..b07f05d5eb8 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/BlockingTask.java
@@ -49,4 +49,11 @@ public interface BlockingTask extends Task {
* task was interrupted.
*/
boolean run(BooleanSupplier supplier);
+
+ /**
+ * Whether the task has been submitted for running (the state of the task
can be waiting for next run etc).
+ *
+ * @return true if the run method has been called.
+ */
+ boolean isRunning();
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
index 3fa8f8abecc..24a0d08aa29 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/task/ForegroundTask.java
@@ -18,6 +18,7 @@ package org.apache.camel.support.task;
import java.time.Duration;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -73,6 +74,7 @@ public class ForegroundTask implements BlockingTask {
private final String name;
private final IterationBudget budget;
private Duration elapsed = Duration.ZERO;
+ private final AtomicBoolean running = new AtomicBoolean();
ForegroundTask(IterationBudget budget, String name) {
this.budget = budget;
@@ -81,6 +83,7 @@ public class ForegroundTask implements BlockingTask {
@Override
public boolean run(BooleanSupplier supplier) {
+ running.set(true);
boolean completed = false;
try {
@@ -93,7 +96,6 @@ public class ForegroundTask implements BlockingTask {
LOG.debug("Task {} is complete after {} iterations and it
is ready to continue",
name, budget.iteration());
completed = true;
-
break;
}
@@ -106,6 +108,7 @@ public class ForegroundTask implements BlockingTask {
Thread.currentThread().interrupt();
} finally {
elapsed = budget.elapsed();
+ running.set(false);
}
return completed;
@@ -120,6 +123,7 @@ public class ForegroundTask implements BlockingTask {
* @return An optional with the result
*/
public <T> Optional<T> run(Supplier<T> supplier, Predicate<T> predicate) {
+ running.set(true);
try {
if (budget.initialDelay() > 0) {
Thread.sleep(budget.initialDelay());
@@ -142,11 +146,17 @@ public class ForegroundTask implements BlockingTask {
Thread.currentThread().interrupt();
} finally {
elapsed = budget.elapsed();
+ running.set(false);
}
return Optional.empty();
}
+ @Override
+ public boolean isRunning() {
+ return running.get();
+ }
+
@Override
public Duration elapsed() {
return elapsed;
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/MasterComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/MasterComponentBuilderFactory.java
index 7d202c99539..4dc16924c10 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/MasterComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/MasterComponentBuilderFactory.java
@@ -126,14 +126,14 @@ public interface MasterComponentBuilderFactory {
* max attempts reached. This option is the maximum number of attempts
* to try.
*
- * The option is a: <code>long</code> type.
+ * The option is a: <code>int</code> type.
*
* Group: advanced
*
* @param backOffMaxAttempts the value to set
* @return the dsl builder
*/
- default MasterComponentBuilder backOffMaxAttempts(long
backOffMaxAttempts) {
+ default MasterComponentBuilder backOffMaxAttempts(int
backOffMaxAttempts) {
doSetProperty("backOffMaxAttempts", backOffMaxAttempts);
return this;
}
@@ -189,7 +189,7 @@ public interface MasterComponentBuilderFactory {
case "bridgeErrorHandler": ((MasterComponent)
component).setBridgeErrorHandler((boolean) value); return true;
case "autowiredEnabled": ((MasterComponent)
component).setAutowiredEnabled((boolean) value); return true;
case "backOffDelay": ((MasterComponent)
component).setBackOffDelay((long) value); return true;
- case "backOffMaxAttempts": ((MasterComponent)
component).setBackOffMaxAttempts((long) value); return true;
+ case "backOffMaxAttempts": ((MasterComponent)
component).setBackOffMaxAttempts((int) value); return true;
case "service": ((MasterComponent)
component).setService((org.apache.camel.cluster.CamelClusterService) value);
return true;
case "serviceSelector": ((MasterComponent)
component).setServiceSelector((org.apache.camel.cluster.CamelClusterService.Selector)
value); return true;
default: return false;
