You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2024/07/03 11:16:42 UTC
(camel) 01/01: CAMEL-15903: camel-master - Should retry starting consumer on leadership taken.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master-retry
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 89c396d0fcdbcace29dc6530ce25acf74fef2495
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Jul 3 13:16:20 2024 +0200
CAMEL-15903: camel-master - Should retry starting consumer on leadership taken.
---
.../master/MasterComponentConfigurer.java | 12 +++++
.../org/apache/camel/component/master/master.json | 6 ++-
.../camel/component/master/MasterComponent.java | 57 ++++++++++++++++++++++
.../camel/component/master/MasterConsumer.java | 55 +++++++++++++--------
.../camel/component/master/MasterEndpoint.java | 5 ++
5 files changed, 114 insertions(+), 21 deletions(-)
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 fc783ff5d65..81a7ecfe5e7 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
@@ -25,6 +25,10 @@ public class MasterComponentConfigurer extends PropertyConfigurerSupport impleme
switch (ignoreCase ? name.toLowerCase() : name) {
case "autowiredenabled":
case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true;
+ 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 "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;
@@ -39,6 +43,10 @@ public class MasterComponentConfigurer extends PropertyConfigurerSupport impleme
switch (ignoreCase ? name.toLowerCase() : name) {
case "autowiredenabled":
case "autowiredEnabled": return boolean.class;
+ case "backoffdelay":
+ case "backOffDelay": return long.class;
+ case "backoffmaxattempts":
+ case "backOffMaxAttempts": return long.class;
case "bridgeerrorhandler":
case "bridgeErrorHandler": return boolean.class;
case "service": return org.apache.camel.cluster.CamelClusterService.class;
@@ -54,6 +62,10 @@ public class MasterComponentConfigurer extends PropertyConfigurerSupport impleme
switch (ignoreCase ? name.toLowerCase() : name) {
case "autowiredenabled":
case "autowiredEnabled": return target.isAutowiredEnabled();
+ case "backoffdelay":
+ case "backOffDelay": return target.getBackOffDelay();
+ case "backoffmaxattempts":
+ case "backOffMaxAttempts": return target.getBackOffMaxAttempts();
case "bridgeerrorhandler":
case "bridgeErrorHandler": return target.isBridgeErrorHandler();
case "service": return target.getService();
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 55dbb1f444e..5e55e0f7e63 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
@@ -25,8 +25,10 @@
"componentProperties": {
"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 [...]
- "service": { "index": 2, "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": 3, "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." }
+ "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." },
+ "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." }
},
"properties": {
"namespace": { "index": 0, "kind": "path", "displayName": "Namespace", "group": "consumer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The name of the cluster namespace 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 1e1e7e5b322..c986abde42e 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,6 +17,7 @@
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;
@@ -37,10 +38,17 @@ import org.apache.camel.util.StringHelper;
*/
@Component("master")
public class MasterComponent extends DefaultComponent {
+
@Metadata(label = "advanced")
private CamelClusterService service;
@Metadata(label = "advanced")
private CamelClusterService.Selector serviceSelector;
+ @Metadata(label = "advanced")
+ private long backOffDelay = 5000;
+ @Metadata(label = "advanced")
+ private long backOffMaxAttempts = 10;
+
+ private ScheduledExecutorService backOffThreadPool;
public MasterComponent() {
this(null);
@@ -103,6 +111,38 @@ public class MasterComponent extends DefaultComponent {
this.serviceSelector = serviceSelector;
}
+ public ScheduledExecutorService getBackOffThreadPool() {
+ return backOffThreadPool;
+ }
+
+ public long getBackOffDelay() {
+ return backOffDelay;
+ }
+
+ /**
+ * 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.
+ */
+ public void setBackOffDelay(long backOffDelay) {
+ this.backOffDelay = backOffDelay;
+ }
+
+ public long getBackOffMaxAttempts() {
+ return backOffMaxAttempts;
+ }
+
+ /**
+ * 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.
+ */
+ public void setBackOffMaxAttempts(long backOffMaxAttempts) {
+ this.backOffMaxAttempts = backOffMaxAttempts;
+ }
+
@Override
protected void doInit() throws Exception {
CamelContext context = getCamelContext();
@@ -114,4 +154,21 @@ public class MasterComponent extends DefaultComponent {
}
}
+ @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 2204e82e322..a34adeaa5b8 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
@@ -33,6 +33,8 @@ import org.apache.camel.resume.ResumeStrategy;
import org.apache.camel.support.DefaultConsumer;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +43,7 @@ import org.slf4j.LoggerFactory;
*/
@ManagedResource(description = "Managed Master Consumer")
public class MasterConsumer extends DefaultConsumer implements ResumeAware<ResumeStrategy> {
- private static final transient Logger LOG = LoggerFactory.getLogger(MasterConsumer.class);
+ private static final Logger LOG = LoggerFactory.getLogger(MasterConsumer.class);
private final CamelClusterService clusterService;
private final MasterEndpoint masterEndpoint;
@@ -133,31 +135,46 @@ public class MasterConsumer extends DefaultConsumer implements ResumeAware<Resum
return;
}
- delegatedConsumer = delegatedEndpoint.createConsumer(processor);
- if (delegatedConsumer instanceof StartupListener) {
- getEndpoint().getCamelContext().addStartupListener((StartupListener) delegatedConsumer);
- }
+ // start consumer using background task up till X attempts
+ long delay = masterEndpoint.getComponent().getBackOffDelay();
+ long max = masterEndpoint.getComponent().getBackOffMaxAttempts();
- if (delegatedConsumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) {
- LOG.info("Setting up the resume adapter for the resume strategy in the delegated consumer");
- ResumeAdapter resumeAdapter
- = AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer, resumeStrategy);
- resumeStrategy.setAdapter(resumeAdapter);
+ BackOffTimer timer = new BackOffTimer(masterEndpoint.getComponent().getBackOffThreadPool());
+ timer.schedule(BackOff.builder().delay(delay).maxAttempts(max).build(), task -> {
+ LOG.info("Leadership taken. Attempt #{} to start consumer: {}", task.getCurrentAttempts(),
+ delegatedEndpoint);
- LOG.info("Setting up the resume strategy for the delegated consumer");
- resumeAwareConsumer.setResumeStrategy(resumeStrategy);
- }
+ if (delegatedConsumer == null) {
+ delegatedConsumer = delegatedEndpoint.createConsumer(processor);
+ if (delegatedConsumer instanceof StartupListener) {
+ getEndpoint().getCamelContext().addStartupListener((StartupListener) delegatedConsumer);
+ }
+ if (delegatedConsumer instanceof ResumeAware resumeAwareConsumer && resumeStrategy != null) {
+ LOG.debug("Setting up the resume adapter for the resume strategy in consumer");
+ ResumeAdapter resumeAdapter
+ = AdapterHelper.eval(clusterService.getCamelContext(), resumeAwareConsumer,
+ resumeStrategy);
+ resumeStrategy.setAdapter(resumeAdapter);
+
+ LOG.debug("Setting up the resume strategy for consumer");
+ resumeAwareConsumer.setResumeStrategy(resumeStrategy);
+ }
+ }
- ServiceHelper.startService(delegatedEndpoint, delegatedConsumer);
+ ServiceHelper.startService(delegatedEndpoint, delegatedConsumer);
- LOG.info("Leadership taken. Consumer started: {}", delegatedEndpoint);
+ LOG.info("Leadership taken. Consumer started: {}", delegatedEndpoint);
+ return true;
+ });
}
private synchronized void onLeadershipLost() {
- ServiceHelper.stopAndShutdownServices(delegatedConsumer, delegatedEndpoint);
-
- delegatedConsumer = null;
-
+ LOG.debug("Leadership lost. Stopping consumer: {}", delegatedEndpoint);
+ try {
+ ServiceHelper.stopAndShutdownServices(delegatedConsumer, delegatedEndpoint);
+ } finally {
+ delegatedConsumer = null;
+ }
LOG.info("Leadership lost. Consumer stopped: {}", delegatedEndpoint);
}
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 428632779bf..34a22565066 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
@@ -63,6 +63,11 @@ public class MasterEndpoint extends DefaultEndpoint implements DelegateEndpoint
this.delegateEndpoint = getCamelContext().getEndpoint(delegateUri);
}
+ @Override
+ public MasterComponent getComponent() {
+ return (MasterComponent) super.getComponent();
+ }
+
@Override
public Producer createProducer() throws Exception {
throw new UnsupportedOperationException("Cannot produce from this endpoint");