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 2020/05/08 20:19:40 UTC
[camel] branch master updated (6c65c95 -> a8b1f96)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git.
from 6c65c95 CAMEL-15031: Accept FQHN for splunk-hec endpoint (#3811)
new aee36d9 CAMEL-13535: Camel main - Allow to configure supervising route controller
new 704205f CAMEL-13535: Camel main - Allow to configure supervising route controller
new 5cd2ac4 CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
new e2d6f45 CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
new a8b1f96 CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/camel/ExtendedCamelContext.java | 11 ++
.../camel/spi/SupervisingRouteController.java | 84 +++++++++
.../camel/impl/engine/AbstractCamelContext.java | 27 ++-
...java => DefaultSupervisingRouteController.java} | 198 +++++++++++----------
.../camel/impl/engine/SimpleCamelContext.java | 7 +-
.../engine/SupervisingRouteControllerFilters.java | 8 +-
.../camel/impl/ExtendedCamelContextConfigurer.java | 5 +
.../camel/impl/lw/LightweightCamelContext.java | 11 ++
.../impl/lw/LightweightRuntimeCamelContext.java | 13 ++
.../MainConfigurationPropertiesConfigurer.java | 40 +++++
.../camel-main-configuration-metadata.json | 8 +
.../camel/main/DefaultConfigurationConfigurer.java | 30 ++++
.../camel/main/DefaultConfigurationProperties.java | 180 +++++++++++++++++++
.../main/MainSupervisingRouteControllerTest.java | 118 ++++++++++++
.../org/apache/camel/util/backoff/BackOff.java | 14 +-
15 files changed, 646 insertions(+), 108 deletions(-)
create mode 100644 core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
rename core/camel-base/src/main/java/org/apache/camel/impl/engine/{SupervisingRouteController.java => DefaultSupervisingRouteController.java} (78%)
create mode 100644 core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
[camel] 02/05: CAMEL-13535: Camel main - Allow to configure
supervising route controller
Posted by da...@apache.org.
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 704205fb1dedae03e58626b22e969244a462e3ec
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri May 8 16:55:39 2020 +0200
CAMEL-13535: Camel main - Allow to configure supervising route controller
---
.../org/apache/camel/impl/engine/SimpleCamelContext.java | 1 -
.../camel/impl/engine/SupervisingRouteController.java | 16 +++++++++-------
2 files changed, 9 insertions(+), 8 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
index 4bddb7d..4c713e6 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
@@ -222,7 +222,6 @@ public class SimpleCamelContext extends AbstractCamelContext {
@Override
protected RouteController createRouteController() {
- // TODO:
return new DefaultRouteController(getCamelContextReference());
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
index ea990ec..c280db9 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
@@ -72,7 +72,7 @@ public class SupervisingRouteController extends DefaultRouteController {
private final AtomicInteger routeCount;
private final List<Filter> filters;
private final Set<RouteHolder> routes;
- private final CamelContextStartupListener listener;
+ private CamelContextStartupListener listener;
private final RouteManager routeManager;
private BackOffTimer timer;
private ScheduledExecutorService executorService;
@@ -91,12 +91,6 @@ public class SupervisingRouteController extends DefaultRouteController {
this.backOffConfigurations = new HashMap<>();
this.initialDelay = Duration.ofMillis(0);
- try {
- this.listener = new CamelContextStartupListener();
- this.listener.start();
- } catch (Exception e) {
- throw RuntimeCamelException.wrapRuntimeException(e);
- }
}
// *********************************
@@ -201,11 +195,19 @@ public class SupervisingRouteController extends DefaultRouteController {
@Override
protected void doInit() throws Exception {
+ this.listener = new CamelContextStartupListener();
+
CamelContext context = getCamelContext();
context.setAutoStartup(false);
context.addRoutePolicyFactory(new ManagedRoutePolicyFactory());
context.addStartupListener(this.listener);
context.getManagementStrategy().addEventNotifier(this.listener);
+
+ try {
+ this.listener.start();
+ } catch (Exception e) {
+ throw RuntimeCamelException.wrapRuntimeException(e);
+ }
}
@Override
[camel] 04/05: CAMEL-15032: SupervisingRouteController - Make it
simpler and unify across runtimes
Posted by da...@apache.org.
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 e2d6f450658684884aac7a6ce3f92cbcc4bbb926
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri May 8 17:33:37 2020 +0200
CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
---
.../org/apache/camel/ExtendedCamelContext.java | 11 ++++
.../camel/spi/SupervisingRouteController.java | 10 ++++
.../camel/impl/engine/AbstractCamelContext.java | 25 +++++++++-
.../engine/DefaultSupervisingRouteController.java | 58 ++++++++++++++--------
.../camel/impl/engine/SimpleCamelContext.java | 6 +++
.../camel/impl/lw/LightweightCamelContext.java | 11 ++++
.../impl/lw/LightweightRuntimeCamelContext.java | 13 +++++
.../camel/main/DefaultConfigurationConfigurer.java | 9 ++--
.../camel/main/DefaultConfigurationProperties.java | 15 ++++++
.../main/MainSupervisingRouteControllerTest.java | 10 +++-
10 files changed, 140 insertions(+), 28 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java b/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
index b58fb7f..b75c91e 100644
--- a/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
+++ b/core/camel-api/src/main/java/org/apache/camel/ExtendedCamelContext.java
@@ -56,6 +56,7 @@ import org.apache.camel.spi.Registry;
import org.apache.camel.spi.RestBindingJaxbDataFormatFactory;
import org.apache.camel.spi.RouteController;
import org.apache.camel.spi.RouteStartupOrder;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.UnitOfWorkFactory;
import org.apache.camel.spi.XMLRoutesDefinitionLoader;
@@ -554,6 +555,16 @@ public interface ExtendedCamelContext extends CamelContext {
*/
void setConfigurerResolver(ConfigurerResolver configurerResolver);
+ /**
+ * Gets the {@link SupervisingRouteController} to use.
+ */
+ SupervisingRouteController getSupervisingRouteController();
+
+ /**
+ * Sets the {@link SupervisingRouteController} to use.
+ */
+ void setSupervisingRouteController(SupervisingRouteController supervisingRouteController);
+
RouteController getInternalRouteController();
void addRoute(Route route);
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java b/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
index 4b6e936..2f6b50d 100644
--- a/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
@@ -24,6 +24,16 @@ package org.apache.camel.spi;
*/
public interface SupervisingRouteController extends RouteController {
+ int getThreadPoolSize();
+
+ /**
+ * The number of threads used by the scheduled thread pool that are used for restarting
+ * routes. The pool uses 1 thread by default, but you can increase this to allow the controller
+ * to concurrently attempt to restart multiple routes in case more than one route has problems
+ * starting.
+ */
+ void setThreadPoolSize(int threadPoolSize);
+
long getInitialDelay();
/**
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index a0a63ef..c9c6378 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -129,7 +129,6 @@ import org.apache.camel.spi.ProcessorFactory;
import org.apache.camel.spi.PropertiesComponent;
import org.apache.camel.spi.ReactiveExecutor;
import org.apache.camel.spi.Registry;
-import org.apache.camel.spi.ReifierStrategy;
import org.apache.camel.spi.RestBindingJaxbDataFormatFactory;
import org.apache.camel.spi.RestConfiguration;
import org.apache.camel.spi.RestRegistry;
@@ -141,6 +140,7 @@ import org.apache.camel.spi.RouteStartupOrder;
import org.apache.camel.spi.RuntimeEndpointRegistry;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.TransformerRegistry;
@@ -277,6 +277,7 @@ public abstract class AbstractCamelContext extends BaseService
private volatile UuidGenerator uuidGenerator;
private volatile UnitOfWorkFactory unitOfWorkFactory;
private volatile RouteController routeController;
+ private volatile SupervisingRouteController supervisingRouteController;
private volatile ScheduledExecutorService errorHandlerExecutorService;
private volatile BeanIntrospection beanIntrospection;
private volatile Tracer tracer;
@@ -338,6 +339,7 @@ public abstract class AbstractCamelContext extends BaseService
this.startupListeners.add(deferStartupListener);
setDefaultExtension(HealthCheckRegistry.class, this::createHealthCheckRegistry);
+ setDefaultExtension(RouteController.class, this::createRouteController);
if (build) {
try {
@@ -1097,6 +1099,23 @@ public abstract class AbstractCamelContext extends BaseService
}
@Override
+ public SupervisingRouteController getSupervisingRouteController() {
+ if (supervisingRouteController == null) {
+ synchronized (lock) {
+ if (supervisingRouteController == null) {
+ setSupervisingRouteController(createSupervisingRouteController());
+ }
+ }
+ }
+ return supervisingRouteController;
+ }
+
+ @Override
+ public void setSupervisingRouteController(SupervisingRouteController supervisingRouteController) {
+ this.supervisingRouteController = supervisingRouteController;
+ }
+
+ @Override
public List<RouteStartupOrder> getRouteStartupOrder() {
return routeStartupOrder;
}
@@ -2504,7 +2523,7 @@ public abstract class AbstractCamelContext extends BaseService
@Override
public void doInit() throws Exception {
// Start the route controller
- getRouteController();
+ this.routeController = getExtension(RouteController.class);
ServiceHelper.initService(this.routeController);
// optimize - before starting routes lets check if event notifications is possible
@@ -4511,6 +4530,8 @@ public abstract class AbstractCamelContext extends BaseService
protected abstract RouteController createRouteController();
+ protected abstract SupervisingRouteController createSupervisingRouteController();
+
protected abstract ShutdownStrategy createShutdownStrategy();
protected abstract PackageScanClassResolver createPackageScanClassResolver();
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
index fd5d079..cfdca87 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
@@ -64,23 +64,24 @@ import org.slf4j.LoggerFactory;
*/
public class DefaultSupervisingRouteController extends DefaultRouteController implements SupervisingRouteController {
- private static final Logger LOGGER = LoggerFactory.getLogger(DefaultSupervisingRouteController.class);
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultSupervisingRouteController.class);
private final Object lock;
private final AtomicBoolean contextStarted;
private final AtomicInteger routeCount;
private final List<Filter> filters;
private final Set<RouteHolder> routes;
- private CamelContextStartupListener listener;
private final RouteManager routeManager;
- private BackOffTimer timer;
- private ScheduledExecutorService executorService;
+ private volatile CamelContextStartupListener listener;
+ private volatile BackOffTimer timer;
+ private volatile ScheduledExecutorService executorService;
+ private volatile BackOff backOff;
+ private int threadPoolSize = 1;
private long initialDelay;
private long backOffDelay = 2000;
private long backOffMaxDelay;
private long backOffMaxElapsedTime;
private long backOffMaxAttempts;
private double backOffMultiplier = 1.0d;
- private volatile BackOff backOff;
public DefaultSupervisingRouteController() {
this.lock = new Object();
@@ -95,6 +96,16 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
// Properties
// *********************************
+ @Override
+ public int getThreadPoolSize() {
+ return threadPoolSize;
+ }
+
+ @Override
+ public void setThreadPoolSize(int threadPoolSize) {
+ this.threadPoolSize = threadPoolSize;
+ }
+
public long getInitialDelay() {
return initialDelay;
}
@@ -165,10 +176,6 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
return Collections.unmodifiableList(filters);
}
- public Optional<BackOffTimer.Task> getBackOffContext(String id) {
- return routeManager.getBackOffContext(id);
- }
-
protected BackOff getBackOff(String id) {
// currently all routes use the same backoff
return backOff;
@@ -204,7 +211,11 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
@Override
protected void doStart() throws Exception {
CamelContext context = getCamelContext();
- executorService = context.getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "SupervisingRouteController");
+ if (threadPoolSize == 1) {
+ executorService = context.getExecutorServiceManager().newSingleThreadScheduledExecutor(this, "SupervisingRouteController");
+ } else {
+ executorService = context.getExecutorServiceManager().newScheduledThreadPool(this, "SupervisingRouteController", threadPoolSize);
+ }
timer = new BackOffTimer(executorService);
}
@@ -342,7 +353,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
routeManager.release(route);
}
- LOGGER.debug("Route {} has been requested to stop", route.getId());
+ LOG.debug("Route {} has been requested to stop", route.getId());
// Mark the route as un-managed
route.get().setRouteController(null);
@@ -399,7 +410,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
}
}
- LOGGER.info("Total managed routes: {} of which {} successfully started and {} re-starting",
+ LOG.info("Total managed routes: {} of which {} successfully started and {} re-starting",
routes.size(),
routes.stream().filter(r -> r.getStatus() == ServiceStatus.Started).count(),
routeManager.routes.size()
@@ -430,13 +441,16 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
logger.info("Start supervising route: {} with back-off: {}", r.getId(), backOff);
BackOffTimer.Task task = timer.schedule(backOff, context -> {
+ final BackOffTimer.Task state = getBackOffContext(r.getId()).orElse(null);
+ long attempt = state != null ? state.getCurrentAttempts() : 0;
try {
- long attempt = getBackOffContext(r.getId()).map(BackOffTimer.Task::getCurrentAttempts).orElse(0L);
logger.info("Restarting route: {} attempt: {}", r.getId(), attempt);
-
doStartRoute(r, false, rx -> DefaultSupervisingRouteController.super.startRoute(rx.getId()));
return false;
} catch (Exception e) {
+ String cause = e.getClass().getName() + ": " + e.getMessage();
+ logger.info("Failed restarting route: {} attempt: {} due: {} (stacktrace in debug log level)", r.getId(), attempt, cause);
+ logger.debug(" Error restarting route caused by: " + e.getMessage(), e);
return true;
}
});
@@ -453,7 +467,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
final boolean stopped = status.isStopped() || status.isStopping();
if (backOffTask != null && backOffTask.getStatus() == BackOffTimer.Task.Status.Exhausted && stopped) {
- LOGGER.warn("Restarting route: {} is exhausted after {} attempts. No more attempts will be made"
+ LOG.warn("Restarting route: {} is exhausted after {} attempts. No more attempts will be made"
+ " and the route is no longer supervised by this route controller and remains as stopped.",
route.getId(), backOffTask.getCurrentAttempts() - 1);
r.get().setRouteController(null);
@@ -472,7 +486,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
boolean release(RouteHolder route) {
BackOffTimer.Task task = routes.remove(route);
if (task != null) {
- LOGGER.info("Cancel restart task for route {}", route.getId());
+ LOG.info("Cancel restart task for route {}", route.getId());
task.cancel();
}
@@ -584,7 +598,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
@Override
public void onInit(Route route) {
if (!route.isAutoStartup()) {
- LOGGER.info("Route: {} will not be supervised (Reason: has explicit auto-startup flag set to false)", route.getId());
+ LOG.info("Route: {} will not be supervised (Reason: has explicit auto-startup flag set to false)", route.getId());
return;
}
@@ -592,7 +606,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
FilterResult result = filter.apply(route);
if (!result.supervised()) {
- LOGGER.info("Route: {} will not be supervised (Reason: {})", route.getId(), result.reason());
+ LOG.info("Route: {} will not be supervised (Reason: {})", route.getId(), result.reason());
return;
}
}
@@ -603,17 +617,17 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
holder.get().setAutoStartup(false);
if (contextStarted.get()) {
- LOGGER.debug("Context is already started: attempt to start route {}", route.getId());
+ LOG.debug("Context is already started: attempt to start route {}", route.getId());
// Eventually delay the startup of the route a later time
if (initialDelay > 0) {
- LOGGER.debug("Route {} will be started in {} millis", holder.getId(), initialDelay);
+ LOG.debug("Route {} will be started in {} millis", holder.getId(), initialDelay);
executorService.schedule(() -> startRoute(holder), initialDelay, TimeUnit.MILLISECONDS);
} else {
startRoute(holder);
}
} else {
- LOGGER.debug("CamelContext is not yet started. Deferring staring route: {}", holder.getId());
+ LOG.debug("CamelContext is not yet started. Deferring staring route: {}", holder.getId());
}
}
}
@@ -665,7 +679,7 @@ public class DefaultSupervisingRouteController extends DefaultRouteController im
// Eventually delay the startup of the routes a later time
if (initialDelay > 0) {
- LOGGER.debug("Routes will be started in {} millis", initialDelay);
+ LOG.debug("Routes will be started in {} millis", initialDelay);
executorService.schedule(DefaultSupervisingRouteController.this::startRoutes, initialDelay, TimeUnit.MILLISECONDS);
} else {
startRoutes();
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
index 4c713e6..aae67d7 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SimpleCamelContext.java
@@ -68,6 +68,7 @@ import org.apache.camel.spi.RestRegistryFactory;
import org.apache.camel.spi.RouteController;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.TransformerRegistry;
import org.apache.camel.spi.TypeConverterRegistry;
@@ -226,6 +227,11 @@ public class SimpleCamelContext extends AbstractCamelContext {
}
@Override
+ protected SupervisingRouteController createSupervisingRouteController() {
+ return new DefaultSupervisingRouteController();
+ }
+
+ @Override
protected ShutdownStrategy createShutdownStrategy() {
return new DefaultShutdownStrategy(getCamelContextReference());
}
diff --git a/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightCamelContext.java b/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightCamelContext.java
index 9aff7dd..c8c8493 100644
--- a/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightCamelContext.java
+++ b/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightCamelContext.java
@@ -121,6 +121,7 @@ import org.apache.camel.spi.RouteStartupOrder;
import org.apache.camel.spi.RuntimeEndpointRegistry;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.TransformerRegistry;
@@ -497,6 +498,16 @@ public class LightweightCamelContext implements ExtendedCamelContext, CatalogCam
}
@Override
+ public void setSupervisingRouteController(SupervisingRouteController supervisingRouteController) {
+ delegate.adapt(ExtendedCamelContext.class).setSupervisingRouteController(supervisingRouteController);
+ }
+
+ @Override
+ public SupervisingRouteController getSupervisingRouteController() {
+ return delegate.adapt(ExtendedCamelContext.class).getSupervisingRouteController();
+ }
+
+ @Override
public List<Route> getRoutes() {
return delegate.getRoutes();
}
diff --git a/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightRuntimeCamelContext.java b/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightRuntimeCamelContext.java
index c7843eb..82fd468 100644
--- a/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightRuntimeCamelContext.java
+++ b/core/camel-core-engine/src/main/java/org/apache/camel/impl/lw/LightweightRuntimeCamelContext.java
@@ -117,6 +117,7 @@ import org.apache.camel.spi.RouteStartupOrder;
import org.apache.camel.spi.RuntimeEndpointRegistry;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.Tracer;
import org.apache.camel.spi.Transformer;
import org.apache.camel.spi.TransformerRegistry;
@@ -164,6 +165,7 @@ public class LightweightRuntimeCamelContext implements ExtendedCamelContext, Cat
private final ClassLoader applicationContextClassLoader;
private final UnitOfWorkFactory unitOfWorkFactory;
private final RouteController routeController;
+ private final SupervisingRouteController supervisingRouteController;
private final InflightRepository inflightRepository;
private final Injector injector;
private final ClassResolver classResolver;
@@ -208,6 +210,7 @@ public class LightweightRuntimeCamelContext implements ExtendedCamelContext, Cat
applicationContextClassLoader = context.getApplicationContextClassLoader();
unitOfWorkFactory = context.adapt(ExtendedCamelContext.class).getUnitOfWorkFactory();
routeController = context.getRouteController();
+ supervisingRouteController = context.adapt(ExtendedCamelContext.class).getSupervisingRouteController();
inflightRepository = context.getInflightRepository();
globalOptions = context.getGlobalOptions();
injector = context.getInjector();
@@ -692,6 +695,11 @@ public class LightweightRuntimeCamelContext implements ExtendedCamelContext, Cat
}
@Override
+ public SupervisingRouteController getSupervisingRouteController() {
+ return supervisingRouteController;
+ }
+
+ @Override
public List<Route> getRoutes() {
return routes;
}
@@ -1606,6 +1614,11 @@ public class LightweightRuntimeCamelContext implements ExtendedCamelContext, Cat
}
@Override
+ public void setSupervisingRouteController(SupervisingRouteController supervisingRouteController) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
public void addRoutes(RoutesBuilder builder) throws Exception {
throw new UnsupportedOperationException();
}
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
index 16cbb20..24dc774 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
@@ -31,7 +31,6 @@ import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.health.HealthCheckRegistry;
import org.apache.camel.health.HealthCheckRepository;
import org.apache.camel.health.HealthCheckService;
-import org.apache.camel.impl.engine.DefaultSupervisingRouteController;
import org.apache.camel.model.Model;
import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
@@ -60,13 +59,13 @@ import org.apache.camel.spi.RoutePolicyFactory;
import org.apache.camel.spi.RuntimeEndpointRegistry;
import org.apache.camel.spi.ShutdownStrategy;
import org.apache.camel.spi.StreamCachingStrategy;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.spi.ThreadPoolFactory;
import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.spi.UnitOfWorkFactory;
import org.apache.camel.spi.UuidGenerator;
import org.apache.camel.support.jsse.GlobalSSLContextParametersSupplier;
import org.apache.camel.util.ObjectHelper;
-import org.apache.camel.util.backoff.BackOff;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -181,8 +180,11 @@ public final class DefaultConfigurationConfigurer {
// supervisting route controller
if (config.isRouteControllerEnabled()) {
- DefaultSupervisingRouteController src = camelContext.getExtension(DefaultSupervisingRouteController.class);
+ SupervisingRouteController src = camelContext.adapt(ExtendedCamelContext.class).getSupervisingRouteController();
src.setCamelContext(camelContext);
+ if (config.getRouteControllerThreadPoolSize() > 0) {
+ src.setThreadPoolSize(config.getRouteControllerThreadPoolSize());
+ }
if (config.getRouteControllerBackOffDelay() > 0) {
src.setBackOffDelay(config.getRouteControllerBackOffDelay());
}
@@ -201,6 +203,7 @@ public final class DefaultConfigurationConfigurer {
if (config.getRouteControllerBackOffMultiplier() > 0) {
src.setBackOffMultiplier(config.getRouteControllerBackOffMultiplier());
}
+ // use this as route controller
camelContext.setRouteController(src);
}
}
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
index 0a56ebf..d6597e2 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
@@ -82,6 +82,7 @@ public abstract class DefaultConfigurationProperties<T> {
private String xmlRests = "classpath:camel-rest/*.xml";
private boolean lightweight;
private boolean routeControllerEnabled;
+ private int routeControllerThreadPoolSize;
private long routeControllerInitialDelay;
private long routeControllerBackOffDelay;
private long routeControllerBackOffMaxDelay;
@@ -915,6 +916,20 @@ public abstract class DefaultConfigurationProperties<T> {
this.routeControllerEnabled = routeControllerEnabled;
}
+ public int getRouteControllerThreadPoolSize() {
+ return routeControllerThreadPoolSize;
+ }
+
+ /**
+ * The number of threads used by the route controller scheduled thread pool that are used for restarting
+ * routes. The pool uses 1 thread by default, but you can increase this to allow the controller
+ * to concurrently attempt to restart multiple routes in case more than one route has problems
+ * starting.
+ */
+ public void setRouteControllerThreadPoolSize(int routeControllerThreadPoolSize) {
+ this.routeControllerThreadPoolSize = routeControllerThreadPoolSize;
+ }
+
public long getRouteControllerInitialDelay() {
return routeControllerInitialDelay;
}
diff --git a/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java b/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
index d91699d..fae981a 100644
--- a/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
+++ b/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
@@ -41,6 +41,7 @@ public class MainSupervisingRouteControllerTest extends Assert {
main.configure().setRouteControllerBackOffDelay(250);
main.configure().setRouteControllerBackOffMaxAttempts(3);
main.configure().setRouteControllerInitialDelay(1000);
+ main.configure().setRouteControllerThreadPoolSize(2);
main.start();
@@ -50,11 +51,16 @@ public class MainSupervisingRouteControllerTest extends Assert {
MockEndpoint mock2 = main.getCamelContext().getEndpoint("mock:cheese", MockEndpoint.class);
mock2.expectedMessageCount(0);
- MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock, mock2);
+ MockEndpoint mock3 = main.getCamelContext().getEndpoint("mock:cake", MockEndpoint.class);
+ mock3.expectedMessageCount(0);
+
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock, mock2, mock3);
assertEquals("Started", main.camelContext.getRouteController().getRouteStatus("foo").toString());
// cheese was not able to start
assertEquals("Stopped", main.camelContext.getRouteController().getRouteStatus("cheese").toString());
+ // cake was not able to start
+ assertEquals("Stopped", main.camelContext.getRouteController().getRouteStatus("cake").toString());
main.stop();
}
@@ -67,6 +73,8 @@ public class MainSupervisingRouteControllerTest extends Assert {
from("timer:foo").to("mock:foo").routeId("foo");
from("jms:cheese").to("mock:cheese").routeId("cheese");
+
+ from("jms:cake").to("mock:cake").routeId("cake");
}
}
[camel] 03/05: CAMEL-15032: SupervisingRouteController - Make it
simpler and unify across runtimes
Posted by da...@apache.org.
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 5cd2ac433f98f9c9058c6c7ad5d365e9ec89f281
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri May 8 17:15:35 2020 +0200
CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
---
.../camel/spi/SupervisingRouteController.java | 74 +++++++++++
...java => DefaultSupervisingRouteController.java} | 141 ++++++++++-----------
.../engine/SupervisingRouteControllerFilters.java | 8 +-
.../camel/main/DefaultConfigurationConfigurer.java | 30 ++---
4 files changed, 154 insertions(+), 99 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java b/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
new file mode 100644
index 0000000..4b6e936
--- /dev/null
+++ b/core/camel-api/src/main/java/org/apache/camel/spi/SupervisingRouteController.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.spi;
+
+/**
+ * A supervising capable {@link RouteController} that delays the startup
+ * of the routes after the camel context startup and takes control of starting the routes in a safe manner.
+ * This controller is able to retry starting failing routes, and have various options to configure
+ * settings for backoff between restarting routes.
+ */
+public interface SupervisingRouteController extends RouteController {
+
+ long getInitialDelay();
+
+ /**
+ * Initial delay in milli seconds before the route controller starts, after
+ * CamelContext has been started.
+ */
+ void setInitialDelay(long initialDelay);
+
+ long getBackOffDelay();
+
+ /**
+ * Backoff delay in millis when restarting a route that failed to startup.
+ */
+ void setBackOffDelay(long backOffDelay);
+
+ long getBackOffMaxDelay();
+
+ /**
+ * Backoff maximum delay in millis when restarting a route that failed to startup.
+ */
+ void setBackOffMaxDelay(long backOffMaxDelay);
+
+ long getBackOffMaxElapsedTime();
+
+ /**
+ * Backoff maximum elapsed time in millis, after which the backoff should be considered
+ * exhausted and no more attempts should be made.
+ */
+ void setBackOffMaxElapsedTime(long backOffMaxElapsedTime);
+
+ long getBackOffMaxAttempts();
+
+ /**
+ * Backoff maximum number of attempts to restart a route that failed to startup.
+ * When this threshold has been exceeded then the controller will give up
+ * attempting to restart the route, and the route will remain as stopped.
+ */
+ void setBackOffMaxAttempts(long backOffMaxAttempts);
+
+ double getBackOffMultiplier();
+
+ /**
+ * Backoff multiplier to use for exponential backoff. This is used to extend the delay
+ * between restart attempts.
+ */
+ void setBackOffMultiplier(double backOffMultiplier);
+
+}
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
similarity index 86%
rename from core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
rename to core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
index c280db9..fd5d079 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/DefaultSupervisingRouteController.java
@@ -20,7 +20,6 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -47,6 +46,7 @@ import org.apache.camel.spi.HasId;
import org.apache.camel.spi.RouteController;
import org.apache.camel.spi.RoutePolicy;
import org.apache.camel.spi.RoutePolicyFactory;
+import org.apache.camel.spi.SupervisingRouteController;
import org.apache.camel.support.EventNotifierSupport;
import org.apache.camel.support.RoutePolicySupport;
import org.apache.camel.util.ObjectHelper;
@@ -62,11 +62,9 @@ import org.slf4j.LoggerFactory;
* This controller is able to retry starting failing routes, and have various options to configure
* settings for backoff between restarting routes.
*/
-public class SupervisingRouteController extends DefaultRouteController {
+public class DefaultSupervisingRouteController extends DefaultRouteController implements SupervisingRouteController {
- // TODO: Make SPI interface so we can separate this more nicely
-
- private static final Logger LOGGER = LoggerFactory.getLogger(SupervisingRouteController.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(DefaultSupervisingRouteController.class);
private final Object lock;
private final AtomicBoolean contextStarted;
private final AtomicInteger routeCount;
@@ -76,91 +74,73 @@ public class SupervisingRouteController extends DefaultRouteController {
private final RouteManager routeManager;
private BackOffTimer timer;
private ScheduledExecutorService executorService;
- private BackOff defaultBackOff;
- private Map<String, BackOff> backOffConfigurations;
- private Duration initialDelay;
-
- public SupervisingRouteController() {
+ private long initialDelay;
+ private long backOffDelay = 2000;
+ private long backOffMaxDelay;
+ private long backOffMaxElapsedTime;
+ private long backOffMaxAttempts;
+ private double backOffMultiplier = 1.0d;
+ private volatile BackOff backOff;
+
+ public DefaultSupervisingRouteController() {
this.lock = new Object();
this.contextStarted = new AtomicBoolean(false);
this.filters = new ArrayList<>();
this.routeCount = new AtomicInteger(0);
this.routes = new TreeSet<>();
this.routeManager = new RouteManager();
- this.defaultBackOff = BackOff.builder().build();
- this.backOffConfigurations = new HashMap<>();
- this.initialDelay = Duration.ofMillis(0);
-
}
// *********************************
// Properties
// *********************************
- public BackOff getDefaultBackOff() {
- return defaultBackOff;
+ public long getInitialDelay() {
+ return initialDelay;
}
- /**
- * Sets the default back-off.
- */
- public void setDefaultBackOff(BackOff defaultBackOff) {
- this.defaultBackOff = defaultBackOff;
+ public void setInitialDelay(long initialDelay) {
+ this.initialDelay = initialDelay;
}
- public Map<String, BackOff> getBackOffConfigurations() {
- return backOffConfigurations;
+ public long getBackOffDelay() {
+ return backOffDelay;
}
- /**
- * Set the back-off for the given IDs.
- */
- public void setBackOffConfigurations(Map<String, BackOff> backOffConfigurations) {
- this.backOffConfigurations = backOffConfigurations;
+ public void setBackOffDelay(long backOffDelay) {
+ this.backOffDelay = backOffDelay;
}
- public BackOff getBackOff(String id) {
- return backOffConfigurations.getOrDefault(id, defaultBackOff);
+ public long getBackOffMaxDelay() {
+ return backOffMaxDelay;
}
- /**
- * Sets the back-off to be applied to the given <code>id</code>.
- */
- public void setBackOff(String id, BackOff backOff) {
- backOffConfigurations.put(id, backOff);
+ public void setBackOffMaxDelay(long backOffMaxDelay) {
+ this.backOffMaxDelay = backOffMaxDelay;
}
- public Duration getInitialDelay() {
- return initialDelay;
+ public long getBackOffMaxElapsedTime() {
+ return backOffMaxElapsedTime;
}
- /**
- * Set the amount of time the route controller should wait before to start
- * the routes after the camel context is started or after the route is
- * initialized if the route is created after the camel context is started.
- *
- * @param initialDelay the initial delay.
- */
- public void setInitialDelay(Duration initialDelay) {
- this.initialDelay = initialDelay;
+ public void setBackOffMaxElapsedTime(long backOffMaxElapsedTime) {
+ this.backOffMaxElapsedTime = backOffMaxElapsedTime;
}
- /**
- * #see {@link this#setInitialDelay(Duration)}
- *
- * @param initialDelay the initial delay amount.
- */
- public void setInitialDelay(long initialDelay, TimeUnit initialDelayUnit) {
- this.initialDelay = Duration.ofMillis(initialDelayUnit.toMillis(initialDelay));
+ public long getBackOffMaxAttempts() {
+ return backOffMaxAttempts;
}
- /**
- * #see {@link this#setInitialDelay(Duration)}
- *
- * @param initialDelay the initial delay in milliseconds.
- */
- public void setInitialDelay(long initialDelay) {
- this.initialDelay = Duration.ofMillis(initialDelay);
+ public void setBackOffMaxAttempts(long backOffMaxAttempts) {
+ this.backOffMaxAttempts = backOffMaxAttempts;
+ }
+
+ public double getBackOffMultiplier() {
+ return backOffMultiplier;
+ }
+
+ public void setBackOffMultiplier(double backOffMultiplier) {
+ this.backOffMultiplier = backOffMultiplier;
}
/**
@@ -189,12 +169,23 @@ public class SupervisingRouteController extends DefaultRouteController {
return routeManager.getBackOffContext(id);
}
+ protected BackOff getBackOff(String id) {
+ // currently all routes use the same backoff
+ return backOff;
+ }
+
// *********************************
// Lifecycle
// *********************************
@Override
protected void doInit() throws Exception {
+ this.backOff = new BackOff(
+ Duration.ofMillis(backOffDelay),
+ backOffMaxDelay > 0 ? Duration.ofMillis(backOffMaxDelay) : Duration.ofMillis(Long.MAX_VALUE),
+ backOffMaxElapsedTime > 0 ? Duration.ofMillis(backOffMaxElapsedTime) : Duration.ofMillis(Long.MAX_VALUE),
+ backOffMaxAttempts > 0 ? backOffMaxAttempts : Long.MAX_VALUE,
+ backOffMultiplier);
this.listener = new CamelContextStartupListener();
CamelContext context = getCamelContext();
@@ -351,7 +342,7 @@ public class SupervisingRouteController extends DefaultRouteController {
routeManager.release(route);
}
- LOGGER.info("Route {} has been requested to stop: stop supervising it", route.getId());
+ LOGGER.debug("Route {} has been requested to stop", route.getId());
// Mark the route as un-managed
route.get().setRouteController(null);
@@ -429,7 +420,7 @@ public class SupervisingRouteController extends DefaultRouteController {
}
void start(RouteHolder route) {
- route.get().setRouteController(SupervisingRouteController.this);
+ route.get().setRouteController(DefaultSupervisingRouteController.this);
routes.computeIfAbsent(
route,
@@ -443,7 +434,7 @@ public class SupervisingRouteController extends DefaultRouteController {
long attempt = getBackOffContext(r.getId()).map(BackOffTimer.Task::getCurrentAttempts).orElse(0L);
logger.info("Restarting route: {} attempt: {}", r.getId(), attempt);
- doStartRoute(r, false, rx -> SupervisingRouteController.super.startRoute(rx.getId()));
+ doStartRoute(r, false, rx -> DefaultSupervisingRouteController.super.startRoute(rx.getId()));
return false;
} catch (Exception e) {
return true;
@@ -580,10 +571,10 @@ public class SupervisingRouteController extends DefaultRouteController {
private void startRoute(RouteHolder holder) {
try {
- SupervisingRouteController.this.doStartRoute(
+ DefaultSupervisingRouteController.this.doStartRoute(
holder,
true,
- r -> SupervisingRouteController.super.startRoute(r.getId())
+ r -> DefaultSupervisingRouteController.super.startRoute(r.getId())
);
} catch (Exception e) {
throw new RuntimeCamelException(e);
@@ -593,7 +584,7 @@ public class SupervisingRouteController extends DefaultRouteController {
@Override
public void onInit(Route route) {
if (!route.isAutoStartup()) {
- LOGGER.info("Route {} won't be supervised (reason: has explicit auto-startup flag set to false)", route.getId());
+ LOGGER.info("Route: {} will not be supervised (Reason: has explicit auto-startup flag set to false)", route.getId());
return;
}
@@ -601,28 +592,28 @@ public class SupervisingRouteController extends DefaultRouteController {
FilterResult result = filter.apply(route);
if (!result.supervised()) {
- LOGGER.info("Route {} won't be supervised (reason: {})", route.getId(), result.reason());
+ LOGGER.info("Route: {} will not be supervised (Reason: {})", route.getId(), result.reason());
return;
}
}
RouteHolder holder = new RouteHolder(route, routeCount.incrementAndGet());
if (routes.add(holder)) {
- holder.get().setRouteController(SupervisingRouteController.this);
+ holder.get().setRouteController(DefaultSupervisingRouteController.this);
holder.get().setAutoStartup(false);
if (contextStarted.get()) {
LOGGER.debug("Context is already started: attempt to start route {}", route.getId());
// Eventually delay the startup of the route a later time
- if (initialDelay.toMillis() > 0) {
- LOGGER.debug("Route {} will be started in {}", holder.getId(), initialDelay);
- executorService.schedule(() -> startRoute(holder), initialDelay.toMillis(), TimeUnit.MILLISECONDS);
+ if (initialDelay > 0) {
+ LOGGER.debug("Route {} will be started in {} millis", holder.getId(), initialDelay);
+ executorService.schedule(() -> startRoute(holder), initialDelay, TimeUnit.MILLISECONDS);
} else {
startRoute(holder);
}
} else {
- LOGGER.debug("Context is not yet started: defer route {} start", holder.getId());
+ LOGGER.debug("CamelContext is not yet started. Deferring staring route: {}", holder.getId());
}
}
}
@@ -673,9 +664,9 @@ public class SupervisingRouteController extends DefaultRouteController {
if (contextStarted.compareAndSet(false, true)) {
// Eventually delay the startup of the routes a later time
- if (initialDelay.toMillis() > 0) {
- LOGGER.debug("Routes will be started in {}", initialDelay);
- executorService.schedule(SupervisingRouteController.this::startRoutes, initialDelay.toMillis(), TimeUnit.MILLISECONDS);
+ if (initialDelay > 0) {
+ LOGGER.debug("Routes will be started in {} millis", initialDelay);
+ executorService.schedule(DefaultSupervisingRouteController.this::startRoutes, initialDelay, TimeUnit.MILLISECONDS);
} else {
startRoutes();
}
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteControllerFilters.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteControllerFilters.java
index 8a41238..e531636 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteControllerFilters.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteControllerFilters.java
@@ -24,14 +24,14 @@ import java.util.Set;
import org.apache.camel.Route;
/**
- * A {@link SupervisingRouteController.Filter} which blacklists routes.
+ * A {@link DefaultSupervisingRouteController.Filter} which blacklists routes.
*/
public final class SupervisingRouteControllerFilters {
private SupervisingRouteControllerFilters() {
}
- public static final class BlackList implements SupervisingRouteController.Filter {
+ public static final class BlackList implements DefaultSupervisingRouteController.Filter {
private final Set<String> names;
public BlackList(String name) {
@@ -43,10 +43,10 @@ public final class SupervisingRouteControllerFilters {
}
@Override
- public SupervisingRouteController.FilterResult apply(Route route) {
+ public DefaultSupervisingRouteController.FilterResult apply(Route route) {
boolean supervised = !names.contains(route.getId());
- return new SupervisingRouteController.FilterResult(
+ return new DefaultSupervisingRouteController.FilterResult(
supervised,
supervised ? null : "black-listed"
);
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
index 29a7363..16cbb20 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
@@ -31,7 +31,7 @@ import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.health.HealthCheckRegistry;
import org.apache.camel.health.HealthCheckRepository;
import org.apache.camel.health.HealthCheckService;
-import org.apache.camel.impl.engine.SupervisingRouteController;
+import org.apache.camel.impl.engine.DefaultSupervisingRouteController;
import org.apache.camel.model.Model;
import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
@@ -181,35 +181,25 @@ public final class DefaultConfigurationConfigurer {
// supervisting route controller
if (config.isRouteControllerEnabled()) {
- SupervisingRouteController src = new SupervisingRouteController();
+ DefaultSupervisingRouteController src = camelContext.getExtension(DefaultSupervisingRouteController.class);
src.setCamelContext(camelContext);
+ if (config.getRouteControllerBackOffDelay() > 0) {
+ src.setBackOffDelay(config.getRouteControllerBackOffDelay());
+ }
if (config.getRouteControllerInitialDelay() > 0) {
src.setInitialDelay(config.getRouteControllerInitialDelay());
}
- BackOff.Builder builder = BackOff.builder();
- boolean flag = false;
- if (config.getRouteControllerBackOffDelay() > 0) {
- builder.delay(config.getRouteControllerBackOffDelay());
- flag = true;
+ if (config.getRouteControllerBackOffMaxAttempts() > 0) {
+ src.setBackOffMaxAttempts(config.getRouteControllerBackOffMaxAttempts());
}
if (config.getRouteControllerBackOffMaxDelay() > 0) {
- builder.maxDelay(config.getRouteControllerBackOffMaxDelay());
- flag = true;
- }
- if (config.getRouteControllerBackOffMaxAttempts() > 0) {
- builder.maxAttempts(config.getRouteControllerBackOffMaxAttempts());
- flag = true;
+ src.setBackOffMaxDelay(config.getRouteControllerBackOffDelay());
}
if (config.getRouteControllerBackOffMaxElapsedTime() > 0) {
- builder.maxElapsedTime(config.getRouteControllerBackOffMaxElapsedTime());
- flag = true;
+ src.setBackOffMaxElapsedTime(config.getRouteControllerBackOffMaxElapsedTime());
}
if (config.getRouteControllerBackOffMultiplier() > 0) {
- builder.multiplier(config.getRouteControllerBackOffMultiplier());
- flag = true;
- }
- if (flag) {
- src.setDefaultBackOff(builder.build());
+ src.setBackOffMultiplier(config.getRouteControllerBackOffMultiplier());
}
camelContext.setRouteController(src);
}
[camel] 05/05: CAMEL-15032: SupervisingRouteController - Make it
simpler and unify across runtimes
Posted by da...@apache.org.
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 a8b1f9663ed4698176f996222b038ba883e0d2bc
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri May 8 22:12:04 2020 +0200
CAMEL-15032: SupervisingRouteController - Make it simpler and unify across runtimes
---
.../main/java/org/apache/camel/impl/engine/AbstractCamelContext.java | 4 ++--
.../java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java | 5 +++++
.../org/apache/camel/main/MainConfigurationPropertiesConfigurer.java | 5 +++++
.../resources/META-INF/camel-main-configuration-metadata.json | 1 +
4 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index c9c6378..c175ae3 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -2522,8 +2522,8 @@ public abstract class AbstractCamelContext extends BaseService
@Override
public void doInit() throws Exception {
- // Start the route controller
- this.routeController = getExtension(RouteController.class);
+ // start the route controller
+ this.routeController = getRouteController();
ServiceHelper.initService(this.routeController);
// optimize - before starting routes lets check if event notifications is possible
diff --git a/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java b/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
index 2ba64b3..895c1e1 100644
--- a/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
+++ b/core/camel-core-engine/src/generated/java/org/apache/camel/impl/ExtendedCamelContextConfigurer.java
@@ -131,6 +131,8 @@ public class ExtendedCamelContextConfigurer extends org.apache.camel.support.com
case "StreamCaching": target.setStreamCaching(property(camelContext, java.lang.Boolean.class, value)); return true;
case "streamcachingstrategy":
case "StreamCachingStrategy": target.setStreamCachingStrategy(property(camelContext, org.apache.camel.spi.StreamCachingStrategy.class, value)); return true;
+ case "supervisingroutecontroller":
+ case "SupervisingRouteController": target.setSupervisingRouteController(property(camelContext, org.apache.camel.spi.SupervisingRouteController.class, value)); return true;
case "tracer":
case "Tracer": target.setTracer(property(camelContext, org.apache.camel.spi.Tracer.class, value)); return true;
case "tracing":
@@ -216,6 +218,7 @@ public class ExtendedCamelContextConfigurer extends org.apache.camel.support.com
answer.put("ShutdownStrategy", org.apache.camel.spi.ShutdownStrategy.class);
answer.put("StreamCaching", java.lang.Boolean.class);
answer.put("StreamCachingStrategy", org.apache.camel.spi.StreamCachingStrategy.class);
+ answer.put("SupervisingRouteController", org.apache.camel.spi.SupervisingRouteController.class);
answer.put("Tracer", org.apache.camel.spi.Tracer.class);
answer.put("Tracing", java.lang.Boolean.class);
answer.put("TracingPattern", java.lang.String.class);
@@ -346,6 +349,8 @@ public class ExtendedCamelContextConfigurer extends org.apache.camel.support.com
case "StreamCaching": return target.isStreamCaching();
case "streamcachingstrategy":
case "StreamCachingStrategy": return target.getStreamCachingStrategy();
+ case "supervisingroutecontroller":
+ case "SupervisingRouteController": return target.getSupervisingRouteController();
case "tracer":
case "Tracer": return target.getTracer();
case "tracing":
diff --git a/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java b/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
index f167ac7..b19c4e1 100644
--- a/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
+++ b/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
@@ -119,6 +119,8 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
case "RouteControllerEnabled": target.setRouteControllerEnabled(property(camelContext, boolean.class, value)); return true;
case "routecontrollerinitialdelay":
case "RouteControllerInitialDelay": target.setRouteControllerInitialDelay(property(camelContext, long.class, value)); return true;
+ case "routecontrollerthreadpoolsize":
+ case "RouteControllerThreadPoolSize": target.setRouteControllerThreadPoolSize(property(camelContext, int.class, value)); return true;
case "routefilterexcludepattern":
case "RouteFilterExcludePattern": target.setRouteFilterExcludePattern(property(camelContext, java.lang.String.class, value)); return true;
case "routefilterincludepattern":
@@ -232,6 +234,7 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
answer.put("RouteControllerBackOffMultiplier", double.class);
answer.put("RouteControllerEnabled", boolean.class);
answer.put("RouteControllerInitialDelay", long.class);
+ answer.put("RouteControllerThreadPoolSize", int.class);
answer.put("RouteFilterExcludePattern", java.lang.String.class);
answer.put("RouteFilterIncludePattern", java.lang.String.class);
answer.put("RoutesBuilderClasses", java.lang.String.class);
@@ -367,6 +370,8 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
case "RouteControllerEnabled": return target.isRouteControllerEnabled();
case "routecontrollerinitialdelay":
case "RouteControllerInitialDelay": return target.getRouteControllerInitialDelay();
+ case "routecontrollerthreadpoolsize":
+ case "RouteControllerThreadPoolSize": return target.getRouteControllerThreadPoolSize();
case "routefilterexcludepattern":
case "RouteFilterExcludePattern": return target.getRouteFilterExcludePattern();
case "routefilterincludepattern":
diff --git a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index 5727428..94e1bfe 100644
--- a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++ b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -57,6 +57,7 @@
{ "name": "camel.main.routeControllerBackOffMultiplier", "description": "Backoff multiplier to use for exponential backoff. This is used to extend the delay between restart attempts.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "number", "javaType": "double" },
{ "name": "camel.main.routeControllerEnabled", "description": "To enable using supervising route controller which allows Camel to startup and then the controller takes care of starting the routes in a safe manner. This can be used when you want to startup Camel despite a route may otherwise fail fast during startup and cause Camel to fail to startup as well. By delegating the route startup to the supervising route controller then its manages the startup using a background thread. The [...]
{ "name": "camel.main.routeControllerInitialDelay", "description": "Initial delay in milli seconds before the route controller starts, after CamelContext has been started.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
+ { "name": "camel.main.routeControllerThreadPoolSize", "description": "The number of threads used by the route controller scheduled thread pool that are used for restarting routes. The pool uses 1 thread by default, but you can increase this to allow the controller to concurrently attempt to restart multiple routes in case more than one route has problems starting.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "int" },
{ "name": "camel.main.routeFilterExcludePattern", "description": "Used for filtering routes routes matching the given pattern, which follows the following rules: - Match by route id - Match by route input endpoint uri The matching is using exact match, by wildcard and regular expression as documented by PatternHelper#matchPattern(String,String) . For example to only include routes which starts with foo in their route id's, use: include=foo* And to exclude routes which starts from [...]
{ "name": "camel.main.routeFilterIncludePattern", "description": "Used for filtering routes routes matching the given pattern, which follows the following rules: - Match by route id - Match by route input endpoint uri The matching is using exact match, by wildcard and regular expression as documented by PatternHelper#matchPattern(String,String) . For example to only include routes which starts with foo in their route id's, use: include=foo* And to exclude routes which starts from [...]
{ "name": "camel.main.routesBuilderClasses", "description": "Sets classes names that implement RoutesBuilder .", "sourceType": "org.apache.camel.main.MainConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
[camel] 01/05: CAMEL-13535: Camel main - Allow to configure
supervising route controller
Posted by da...@apache.org.
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 aee36d9381b61926c40cf9d3433f9b1aab8a7313
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri May 8 14:32:17 2020 +0200
CAMEL-13535: Camel main - Allow to configure supervising route controller
---
.../impl/engine/SupervisingRouteController.java | 19 ++-
.../MainConfigurationPropertiesConfigurer.java | 35 +++++
.../camel-main-configuration-metadata.json | 7 +
.../camel/main/DefaultConfigurationConfigurer.java | 37 +++++
.../camel/main/DefaultConfigurationProperties.java | 165 +++++++++++++++++++++
.../main/MainSupervisingRouteControllerTest.java | 110 ++++++++++++++
.../org/apache/camel/util/backoff/BackOff.java | 14 +-
7 files changed, 373 insertions(+), 14 deletions(-)
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
index 890960a..ea990ec 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/SupervisingRouteController.java
@@ -63,6 +63,9 @@ import org.slf4j.LoggerFactory;
* settings for backoff between restarting routes.
*/
public class SupervisingRouteController extends DefaultRouteController {
+
+ // TODO: Make SPI interface so we can separate this more nicely
+
private static final Logger LOGGER = LoggerFactory.getLogger(SupervisingRouteController.class);
private final Object lock;
private final AtomicBoolean contextStarted;
@@ -92,7 +95,7 @@ public class SupervisingRouteController extends DefaultRouteController {
this.listener = new CamelContextStartupListener();
this.listener.start();
} catch (Exception e) {
- throw new RuntimeException(e);
+ throw RuntimeCamelException.wrapRuntimeException(e);
}
}
@@ -169,6 +172,7 @@ public class SupervisingRouteController extends DefaultRouteController {
/**
* Add a filter used to determine the routes to supervise.
*/
+ @Deprecated
public void addFilter(Filter filter) {
this.filters.add(filter);
}
@@ -176,11 +180,13 @@ public class SupervisingRouteController extends DefaultRouteController {
/**
* Sets the filters user to determine the routes to supervise.
*/
+ @Deprecated
public void setFilters(Collection<Filter> filters) {
this.filters.clear();
this.filters.addAll(filters);
}
+ @Deprecated
public Collection<Filter> getFilters() {
return Collections.unmodifiableList(filters);
}
@@ -432,7 +438,8 @@ public class SupervisingRouteController extends DefaultRouteController {
BackOffTimer.Task task = timer.schedule(backOff, context -> {
try {
- logger.info("Try to restart route: {}", r.getId());
+ long attempt = getBackOffContext(r.getId()).map(BackOffTimer.Task::getCurrentAttempts).orElse(0L);
+ logger.info("Restarting route: {} attempt: {}", r.getId(), attempt);
doStartRoute(r, false, rx -> SupervisingRouteController.super.startRoute(rx.getId()));
return false;
@@ -453,7 +460,9 @@ public class SupervisingRouteController extends DefaultRouteController {
final boolean stopped = status.isStopped() || status.isStopping();
if (backOffTask != null && backOffTask.getStatus() == BackOffTimer.Task.Status.Exhausted && stopped) {
- LOGGER.info("Back-off for route {} is exhausted, no more attempts will be made and stop supervising it", route.getId());
+ LOGGER.warn("Restarting route: {} is exhausted after {} attempts. No more attempts will be made"
+ + " and the route is no longer supervised by this route controller and remains as stopped.",
+ route.getId(), backOffTask.getCurrentAttempts() - 1);
r.get().setRouteController(null);
}
}
@@ -601,7 +610,7 @@ public class SupervisingRouteController extends DefaultRouteController {
holder.get().setAutoStartup(false);
if (contextStarted.get()) {
- LOGGER.info("Context is already started: attempt to start route {}", route.getId());
+ LOGGER.debug("Context is already started: attempt to start route {}", route.getId());
// Eventually delay the startup of the route a later time
if (initialDelay.toMillis() > 0) {
@@ -611,7 +620,7 @@ public class SupervisingRouteController extends DefaultRouteController {
startRoute(holder);
}
} else {
- LOGGER.info("Context is not yet started: defer route {} start", holder.getId());
+ LOGGER.debug("Context is not yet started: defer route {} start", holder.getId());
}
}
}
diff --git a/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java b/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
index a06e794..f167ac7 100644
--- a/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
+++ b/core/camel-main/src/generated/java/org/apache/camel/main/MainConfigurationPropertiesConfigurer.java
@@ -105,6 +105,20 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
case "PackageScanRouteBuilders": target.setPackageScanRouteBuilders(property(camelContext, java.lang.String.class, value)); return true;
case "producertemplatecachesize":
case "ProducerTemplateCacheSize": target.setProducerTemplateCacheSize(property(camelContext, int.class, value)); return true;
+ case "routecontrollerbackoffdelay":
+ case "RouteControllerBackOffDelay": target.setRouteControllerBackOffDelay(property(camelContext, long.class, value)); return true;
+ case "routecontrollerbackoffmaxattempts":
+ case "RouteControllerBackOffMaxAttempts": target.setRouteControllerBackOffMaxAttempts(property(camelContext, long.class, value)); return true;
+ case "routecontrollerbackoffmaxdelay":
+ case "RouteControllerBackOffMaxDelay": target.setRouteControllerBackOffMaxDelay(property(camelContext, long.class, value)); return true;
+ case "routecontrollerbackoffmaxelapsedtime":
+ case "RouteControllerBackOffMaxElapsedTime": target.setRouteControllerBackOffMaxElapsedTime(property(camelContext, long.class, value)); return true;
+ case "routecontrollerbackoffmultiplier":
+ case "RouteControllerBackOffMultiplier": target.setRouteControllerBackOffMultiplier(property(camelContext, double.class, value)); return true;
+ case "routecontrollerenabled":
+ case "RouteControllerEnabled": target.setRouteControllerEnabled(property(camelContext, boolean.class, value)); return true;
+ case "routecontrollerinitialdelay":
+ case "RouteControllerInitialDelay": target.setRouteControllerInitialDelay(property(camelContext, long.class, value)); return true;
case "routefilterexcludepattern":
case "RouteFilterExcludePattern": target.setRouteFilterExcludePattern(property(camelContext, java.lang.String.class, value)); return true;
case "routefilterincludepattern":
@@ -211,6 +225,13 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
answer.put("Name", java.lang.String.class);
answer.put("PackageScanRouteBuilders", java.lang.String.class);
answer.put("ProducerTemplateCacheSize", int.class);
+ answer.put("RouteControllerBackOffDelay", long.class);
+ answer.put("RouteControllerBackOffMaxAttempts", long.class);
+ answer.put("RouteControllerBackOffMaxDelay", long.class);
+ answer.put("RouteControllerBackOffMaxElapsedTime", long.class);
+ answer.put("RouteControllerBackOffMultiplier", double.class);
+ answer.put("RouteControllerEnabled", boolean.class);
+ answer.put("RouteControllerInitialDelay", long.class);
answer.put("RouteFilterExcludePattern", java.lang.String.class);
answer.put("RouteFilterIncludePattern", java.lang.String.class);
answer.put("RoutesBuilderClasses", java.lang.String.class);
@@ -332,6 +353,20 @@ public class MainConfigurationPropertiesConfigurer extends org.apache.camel.supp
case "PackageScanRouteBuilders": return target.getPackageScanRouteBuilders();
case "producertemplatecachesize":
case "ProducerTemplateCacheSize": return target.getProducerTemplateCacheSize();
+ case "routecontrollerbackoffdelay":
+ case "RouteControllerBackOffDelay": return target.getRouteControllerBackOffDelay();
+ case "routecontrollerbackoffmaxattempts":
+ case "RouteControllerBackOffMaxAttempts": return target.getRouteControllerBackOffMaxAttempts();
+ case "routecontrollerbackoffmaxdelay":
+ case "RouteControllerBackOffMaxDelay": return target.getRouteControllerBackOffMaxDelay();
+ case "routecontrollerbackoffmaxelapsedtime":
+ case "RouteControllerBackOffMaxElapsedTime": return target.getRouteControllerBackOffMaxElapsedTime();
+ case "routecontrollerbackoffmultiplier":
+ case "RouteControllerBackOffMultiplier": return target.getRouteControllerBackOffMultiplier();
+ case "routecontrollerenabled":
+ case "RouteControllerEnabled": return target.isRouteControllerEnabled();
+ case "routecontrollerinitialdelay":
+ case "RouteControllerInitialDelay": return target.getRouteControllerInitialDelay();
case "routefilterexcludepattern":
case "RouteFilterExcludePattern": return target.getRouteFilterExcludePattern();
case "routefilterincludepattern":
diff --git a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
index ab41e7c..5727428 100644
--- a/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
+++ b/core/camel-main/src/generated/resources/META-INF/camel-main-configuration-metadata.json
@@ -50,6 +50,13 @@
{ "name": "camel.main.name", "description": "Sets the name of the CamelContext.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
{ "name": "camel.main.packageScanRouteBuilders", "description": "Sets package names for scanning for org.apache.camel.builder.RouteBuilder classes as candidates to be included. If you are using Spring Boot then its instead recommended to use Spring Boots component scanning and annotate your route builder classes with Component. In other words only use this for Camel Main in standalone mode.", "sourceType": "org.apache.camel.main.MainConfigurationProperties", "type": "string", "javaTy [...]
{ "name": "camel.main.producerTemplateCacheSize", "description": "Producer template endpoints cache size.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "int", "defaultValue": 1000 },
+ { "name": "camel.main.routeControllerBackOffDelay", "description": "Backoff delay in millis when restarting a route that failed to startup.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
+ { "name": "camel.main.routeControllerBackOffMaxAttempts", "description": "Backoff maximum number of attempts to restart a route that failed to startup. When this threshold has been exceeded then the controller will give up attempting to restart the route, and the route will remain as stopped.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
+ { "name": "camel.main.routeControllerBackOffMaxDelay", "description": "Backoff maximum delay in millis when restarting a route that failed to startup.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
+ { "name": "camel.main.routeControllerBackOffMaxElapsedTime", "description": "Backoff maximum elapsed time in millis, after which the backoff should be considered exhausted and no more attempts should be made.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
+ { "name": "camel.main.routeControllerBackOffMultiplier", "description": "Backoff multiplier to use for exponential backoff. This is used to extend the delay between restart attempts.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "number", "javaType": "double" },
+ { "name": "camel.main.routeControllerEnabled", "description": "To enable using supervising route controller which allows Camel to startup and then the controller takes care of starting the routes in a safe manner. This can be used when you want to startup Camel despite a route may otherwise fail fast during startup and cause Camel to fail to startup as well. By delegating the route startup to the supervising route controller then its manages the startup using a background thread. The [...]
+ { "name": "camel.main.routeControllerInitialDelay", "description": "Initial delay in milli seconds before the route controller starts, after CamelContext has been started.", "sourceType": "org.apache.camel.main.DefaultConfigurationProperties", "type": "integer", "javaType": "long" },
{ "name": "camel.main.routeFilterExcludePattern", "description": "Used for filtering routes routes matching the given pattern, which follows the following rules: - Match by route id - Match by route input endpoint uri The matching is using exact match, by wildcard and regular expression as documented by PatternHelper#matchPattern(String,String) . For example to only include routes which starts with foo in their route id's, use: include=foo* And to exclude routes which starts from [...]
{ "name": "camel.main.routeFilterIncludePattern", "description": "Used for filtering routes routes matching the given pattern, which follows the following rules: - Match by route id - Match by route input endpoint uri The matching is using exact match, by wildcard and regular expression as documented by PatternHelper#matchPattern(String,String) . For example to only include routes which starts with foo in their route id's, use: include=foo* And to exclude routes which starts from [...]
{ "name": "camel.main.routesBuilderClasses", "description": "Sets classes names that implement RoutesBuilder .", "sourceType": "org.apache.camel.main.MainConfigurationProperties", "type": "string", "javaType": "java.lang.String" },
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
index d945450..29a7363 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationConfigurer.java
@@ -31,6 +31,7 @@ import org.apache.camel.cluster.CamelClusterService;
import org.apache.camel.health.HealthCheckRegistry;
import org.apache.camel.health.HealthCheckRepository;
import org.apache.camel.health.HealthCheckService;
+import org.apache.camel.impl.engine.SupervisingRouteController;
import org.apache.camel.model.Model;
import org.apache.camel.processor.interceptor.BacklogTracer;
import org.apache.camel.spi.AsyncProcessorAwaitManager;
@@ -65,6 +66,7 @@ import org.apache.camel.spi.UnitOfWorkFactory;
import org.apache.camel.spi.UuidGenerator;
import org.apache.camel.support.jsse.GlobalSSLContextParametersSupplier;
import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.backoff.BackOff;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -176,6 +178,41 @@ public final class DefaultConfigurationConfigurer {
if (config.getRouteFilterIncludePattern() != null || config.getRouteFilterExcludePattern() != null) {
camelContext.getExtension(Model.class).setRouteFilterPattern(config.getRouteFilterIncludePattern(), config.getRouteFilterExcludePattern());
}
+
+ // supervisting route controller
+ if (config.isRouteControllerEnabled()) {
+ SupervisingRouteController src = new SupervisingRouteController();
+ src.setCamelContext(camelContext);
+ if (config.getRouteControllerInitialDelay() > 0) {
+ src.setInitialDelay(config.getRouteControllerInitialDelay());
+ }
+ BackOff.Builder builder = BackOff.builder();
+ boolean flag = false;
+ if (config.getRouteControllerBackOffDelay() > 0) {
+ builder.delay(config.getRouteControllerBackOffDelay());
+ flag = true;
+ }
+ if (config.getRouteControllerBackOffMaxDelay() > 0) {
+ builder.maxDelay(config.getRouteControllerBackOffMaxDelay());
+ flag = true;
+ }
+ if (config.getRouteControllerBackOffMaxAttempts() > 0) {
+ builder.maxAttempts(config.getRouteControllerBackOffMaxAttempts());
+ flag = true;
+ }
+ if (config.getRouteControllerBackOffMaxElapsedTime() > 0) {
+ builder.maxElapsedTime(config.getRouteControllerBackOffMaxElapsedTime());
+ flag = true;
+ }
+ if (config.getRouteControllerBackOffMultiplier() > 0) {
+ builder.multiplier(config.getRouteControllerBackOffMultiplier());
+ flag = true;
+ }
+ if (flag) {
+ src.setDefaultBackOff(builder.build());
+ }
+ camelContext.setRouteController(src);
+ }
}
/**
diff --git a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
index bd27e5c..0a56ebf 100644
--- a/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
+++ b/core/camel-main/src/main/java/org/apache/camel/main/DefaultConfigurationProperties.java
@@ -81,6 +81,13 @@ public abstract class DefaultConfigurationProperties<T> {
private String xmlRoutes = "classpath:camel/*.xml";
private String xmlRests = "classpath:camel-rest/*.xml";
private boolean lightweight;
+ private boolean routeControllerEnabled;
+ private long routeControllerInitialDelay;
+ private long routeControllerBackOffDelay;
+ private long routeControllerBackOffMaxDelay;
+ private long routeControllerBackOffMaxElapsedTime;
+ private long routeControllerBackOffMaxAttempts;
+ private double routeControllerBackOffMultiplier;
// getter and setters
// --------------------------------------------------------------
@@ -890,6 +897,95 @@ public abstract class DefaultConfigurationProperties<T> {
this.lightweight = lightweight;
}
+ public boolean isRouteControllerEnabled() {
+ return routeControllerEnabled;
+ }
+
+ /**
+ * To enable using supervising route controller which allows Camel to startup
+ * and then the controller takes care of starting the routes in a safe manner.
+ *
+ * This can be used when you want to startup Camel despite a route may otherwise
+ * fail fast during startup and cause Camel to fail to startup as well. By delegating
+ * the route startup to the supervising route controller then its manages the startup
+ * using a background thread. The controller allows to be configured with various
+ * settings to attempt to restart failing routes.
+ */
+ public void setRouteControllerEnabled(boolean routeControllerEnabled) {
+ this.routeControllerEnabled = routeControllerEnabled;
+ }
+
+ public long getRouteControllerInitialDelay() {
+ return routeControllerInitialDelay;
+ }
+
+ /**
+ * Initial delay in milli seconds before the route controller starts, after
+ * CamelContext has been started.
+ */
+ public void setRouteControllerInitialDelay(long routeControllerInitialDelay) {
+ this.routeControllerInitialDelay = routeControllerInitialDelay;
+ }
+
+ public long getRouteControllerBackOffDelay() {
+ return routeControllerBackOffDelay;
+ }
+
+ /**
+ * Backoff delay in millis when restarting a route that failed to startup.
+ */
+ public void setRouteControllerBackOffDelay(long routeControllerBackOffDelay) {
+ this.routeControllerBackOffDelay = routeControllerBackOffDelay;
+ }
+
+ public long getRouteControllerBackOffMaxDelay() {
+ return routeControllerBackOffMaxDelay;
+ }
+
+ /**
+ * Backoff maximum delay in millis when restarting a route that failed to startup.
+ */
+ public void setRouteControllerBackOffMaxDelay(long routeControllerBackOffMaxDelay) {
+ this.routeControllerBackOffMaxDelay = routeControllerBackOffMaxDelay;
+ }
+
+ public long getRouteControllerBackOffMaxElapsedTime() {
+ return routeControllerBackOffMaxElapsedTime;
+ }
+
+ /**
+ * Backoff maximum elapsed time in millis, after which the backoff should be considered
+ * exhausted and no more attempts should be made.
+ */
+ public void setRouteControllerBackOffMaxElapsedTime(long routeControllerBackOffMaxElapsedTime) {
+ this.routeControllerBackOffMaxElapsedTime = routeControllerBackOffMaxElapsedTime;
+ }
+
+ public long getRouteControllerBackOffMaxAttempts() {
+ return routeControllerBackOffMaxAttempts;
+ }
+
+ /**
+ * Backoff maximum number of attempts to restart a route that failed to startup.
+ * When this threshold has been exceeded then the controller will give up
+ * attempting to restart the route, and the route will remain as stopped.
+ */
+ public void setRouteControllerBackOffMaxAttempts(long routeControllerBackOffMaxAttempts) {
+ this.routeControllerBackOffMaxAttempts = routeControllerBackOffMaxAttempts;
+ }
+
+ public double getRouteControllerBackOffMultiplier() {
+ return routeControllerBackOffMultiplier;
+ }
+
+ /**
+ * Backoff multiplier to use for exponential backoff. This is used to extend the delay
+ * between restart attempts.
+ */
+ public void setRouteControllerBackOffMultiplier(double routeControllerBackOffMultiplier) {
+ this.routeControllerBackOffMultiplier = routeControllerBackOffMultiplier;
+ }
+
// fluent builders
// --------------------------------------------------------------
@@ -1537,4 +1633,73 @@ public abstract class DefaultConfigurationProperties<T> {
this.lightweight = lightweight;
return (T) this;
}
+
+ /**
+ * To enable using supervising route controller which allows Camel to startup
+ * and then the controller takes care of starting the routes in a safe manner.
+ *
+ * This can be used when you want to startup Camel despite a route may otherwise
+ * fail fast during startup and cause Camel to fail to startup as well. By delegating
+ * the route startup to the supervising route controller then its manages the startup
+ * using a background thread. The controller allows to be configured with various
+ * settings to attempt to restart failing routes.
+ */
+ public T withRouteControllerEnabled(boolean routeControllerEnabled) {
+ this.routeControllerEnabled = routeControllerEnabled;
+ return (T) this;
+ }
+
+ /**
+ * Initial delay in milli seconds before the route controller starts, after
+ * CamelContext has been started.
+ */
+ public T withRouteControllerInitialDelay(long routeControllerInitialDelay) {
+ this.routeControllerInitialDelay = routeControllerInitialDelay;
+ return (T) this;
+ }
+
+ /**
+ * Backoff delay in millis when restarting a route that failed to startup.
+ */
+ public T withRouteControllerBackOffDelay(long routeControllerBackOffDelay) {
+ this.routeControllerBackOffDelay = routeControllerBackOffDelay;
+ return (T) this;
+ }
+
+ /**
+ * Backoff maximum delay in millis when restarting a route that failed to startup.
+ */
+ public T withRouteControllerBackOffMaxDelay(long routeControllerBackOffMaxDelay) {
+ this.routeControllerBackOffMaxDelay = routeControllerBackOffMaxDelay;
+ return (T) this;
+ }
+
+ /**
+ * Backoff maximum elapsed time in millis, after which the backoff should be considered
+ * exhausted and no more attempts should be made.
+ */
+ public T withRouteControllerBackOffMaxElapsedTime(long routeControllerBackOffMaxElapsedTime) {
+ this.routeControllerBackOffMaxElapsedTime = routeControllerBackOffMaxElapsedTime;
+ return (T) this;
+ }
+
+ /**
+ * Backoff maximum number of attempts to restart a route that failed to startup.
+ * When this threshold has been exceeded then the controller will give up
+ * attempting to restart the route, and the route will remain as stopped.
+ */
+ public T withRouteControllerBackOffMaxAttempts(long routeControllerBackOffMaxAttempts) {
+ this.routeControllerBackOffMaxAttempts = routeControllerBackOffMaxAttempts;
+ return (T) this;
+ }
+
+ /**
+ * Backoff multiplier to use for exponential backoff. This is used to extend the delay
+ * between restart attempts.
+ */
+ public T withRouteControllerBackOffMultiplier(double routeControllerBackOffMultiplier) {
+ this.routeControllerBackOffMultiplier = routeControllerBackOffMultiplier;
+ return (T) this;
+ }
+
}
diff --git a/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java b/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
new file mode 100644
index 0000000..d91699d
--- /dev/null
+++ b/core/camel-main/src/test/java/org/apache/camel/main/MainSupervisingRouteControllerTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.main;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.seda.SedaComponent;
+import org.apache.camel.component.seda.SedaConsumer;
+import org.apache.camel.component.seda.SedaEndpoint;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MainSupervisingRouteControllerTest extends Assert {
+
+ @Test
+ public void testMain() throws Exception {
+ // lets make a simple route
+ Main main = new Main();
+ main.configure().addRoutesBuilder(new MyRoute());
+ main.configure().setRouteControllerEnabled(true);
+ main.configure().setRouteControllerBackOffDelay(250);
+ main.configure().setRouteControllerBackOffMaxAttempts(3);
+ main.configure().setRouteControllerInitialDelay(1000);
+
+ main.start();
+
+ MockEndpoint mock = main.getCamelContext().getEndpoint("mock:foo", MockEndpoint.class);
+ mock.expectedMinimumMessageCount(3);
+
+ MockEndpoint mock2 = main.getCamelContext().getEndpoint("mock:cheese", MockEndpoint.class);
+ mock2.expectedMessageCount(0);
+
+ MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, mock, mock2);
+
+ assertEquals("Started", main.camelContext.getRouteController().getRouteStatus("foo").toString());
+ // cheese was not able to start
+ assertEquals("Stopped", main.camelContext.getRouteController().getRouteStatus("cheese").toString());
+
+ main.stop();
+ }
+
+ private class MyRoute extends RouteBuilder {
+ @Override
+ public void configure() throws Exception {
+ getContext().addComponent("jms", new MyJmsComponent());
+
+ from("timer:foo").to("mock:foo").routeId("foo");
+
+ from("jms:cheese").to("mock:cheese").routeId("cheese");
+ }
+ }
+
+ private class MyJmsComponent extends SedaComponent {
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ return new MyJmsEndpoint();
+ }
+ }
+
+ private class MyJmsEndpoint extends SedaEndpoint {
+
+ public MyJmsEndpoint() {
+ super();
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new MyJmsConsumer(this, processor);
+ }
+
+ @Override
+ protected String createEndpointUri() {
+ return "jms:cheese";
+ }
+ }
+
+ private class MyJmsConsumer extends SedaConsumer {
+
+ public MyJmsConsumer(SedaEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ throw new IllegalArgumentException("Cannot start");
+ }
+ }
+
+}
diff --git a/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOff.java b/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOff.java
index d96fca8..7b6d240 100644
--- a/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOff.java
+++ b/core/camel-util/src/main/java/org/apache/camel/util/backoff/BackOff.java
@@ -17,6 +17,7 @@
package org.apache.camel.util.backoff;
import java.time.Duration;
+import java.time.temporal.TemporalUnit;
import java.util.concurrent.TimeUnit;
import org.apache.camel.util.ObjectHelper;
@@ -52,9 +53,6 @@ public final class BackOff {
// Properties
// *************************************
- /**
- * @return the delay to wait before retry the operation.
- */
public Duration getDelay() {
return delay;
}
@@ -96,8 +94,6 @@ public final class BackOff {
/**
* The maximum number of attempts after which the back-off should be considered
* exhausted and no more attempts should be made.
- *
- * @param maxAttempts
*/
public void setMaxAttempts(Long maxAttempts) {
this.maxAttempts = maxAttempts;
@@ -116,13 +112,13 @@ public final class BackOff {
@Override
public String toString() {
- return "BackOff{"
+ return "BackOff["
+ "delay=" + delay
- + ", maxDelay=" + maxDelay
- + ", maxElapsedTime=" + maxElapsedTime
+ + ", maxDelay=" + (maxDelay != MAX_DURATION ? maxDelay : "")
+ + ", maxElapsedTime=" + (maxElapsedTime != MAX_DURATION ? maxElapsedTime : "")
+ ", maxAttempts=" + maxAttempts
+ ", multiplier=" + multiplier
- + '}';
+ + ']';
}
// *****************************************