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 2019/05/22 10:14:49 UTC
[camel] 02/02: CAMEL-13556: Camel main - Fixed duration idle/max to
work better and shutdown cleanly
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 da5fd3fd2cd010317b50ebabfa207764bf500d3d
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed May 22 12:14:24 2019 +0200
CAMEL-13556: Camel main - Fixed duration idle/max to work better and shutdown cleanly
---
.../camel/impl/engine/AbstractCamelContext.java | 9 ++++
.../camel/main/MainDurationEventNotifier.java | 58 ++++++++++++----------
.../java/org/apache/camel/main/MainSupport.java | 20 +++++---
3 files changed, 53 insertions(+), 34 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 f266eb1..6191b73 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
@@ -3406,9 +3406,18 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
}
}
+ // preserve any existing event notifiers that may have been already added
+ List<EventNotifier> notifiers = null;
+ if (managementStrategy != null) {
+ notifiers = managementStrategy.getEventNotifiers();
+ }
+
log.debug("Setting up management with factory: {}", factory);
try {
ManagementStrategy strategy = factory.create(this, options);
+ if (notifiers != null) {
+ notifiers.forEach(strategy::addEventNotifier);
+ }
LifecycleStrategy lifecycle = factory.createLifecycle(this);
factory.setupManagement(this, strategy, lifecycle);
} catch (Exception e) {
diff --git a/core/camel-core/src/main/java/org/apache/camel/main/MainDurationEventNotifier.java b/core/camel-core/src/main/java/org/apache/camel/main/MainDurationEventNotifier.java
index 4c8bd84..cc7c5bf 100644
--- a/core/camel-core/src/main/java/org/apache/camel/main/MainDurationEventNotifier.java
+++ b/core/camel-core/src/main/java/org/apache/camel/main/MainDurationEventNotifier.java
@@ -48,7 +48,7 @@ public class MainDurationEventNotifier extends EventNotifierSupport {
private volatile int doneMessages;
private volatile StopWatch watch;
- private volatile ScheduledExecutorService executorService;
+ private volatile ScheduledExecutorService idleExecutorService;
public MainDurationEventNotifier(CamelContext camelContext, int maxMessages, long maxIdleSeconds,
AtomicBoolean completed, CountDownLatch latch, boolean stopCamelContext) {
@@ -74,16 +74,12 @@ public class MainDurationEventNotifier extends EventNotifierSupport {
if (result) {
if (completed.compareAndSet(false, true)) {
LOG.info("Duration max messages triggering shutdown of the JVM.");
- try {
- // shutting down CamelContext
- if (stopCamelContext) {
- camelContext.stop();
- }
- } catch (Exception e) {
- LOG.warn("Error during stopping CamelContext. This exception is ignored.", e);
- } finally {
- // trigger stopping the Main
- latch.countDown();
+ // use thread to stop Camel as otherwise we would block current thread
+ Thread thread = camelContext.getExecutorServiceManager().newThread("CamelMainShutdownCamelContext", new ShutdownTask());
+ thread.start();
+ // shutdown idle checker if in use as we are stopping
+ if (idleExecutorService != null) {
+ idleExecutorService.shutdownNow();
}
}
}
@@ -98,7 +94,7 @@ public class MainDurationEventNotifier extends EventNotifierSupport {
@Override
public boolean isEnabled(CamelEvent event) {
- return event instanceof ExchangeCompletedEvent || event instanceof ExchangeFailedEvent;
+ return event instanceof ExchangeCreatedEvent || event instanceof ExchangeCompletedEvent || event instanceof ExchangeFailedEvent;
}
@Override
@@ -114,7 +110,7 @@ public class MainDurationEventNotifier extends EventNotifierSupport {
camelContext.addStartupListener((context, alreadyStarted) -> watch = new StopWatch());
// okay we need to trigger on idle after X period, and therefore we need a background task that checks this
- executorService = Executors.newSingleThreadScheduledExecutor();
+ idleExecutorService = Executors.newSingleThreadScheduledExecutor();
Runnable task = () -> {
if (watch == null) {
// camel has not been started yet
@@ -135,23 +131,31 @@ public class MainDurationEventNotifier extends EventNotifierSupport {
if (result) {
if (completed.compareAndSet(false, true)) {
LOG.info("Duration max idle triggering shutdown of the JVM.");
- try {
- // shutting down CamelContext
- if (stopCamelContext) {
- camelContext.stop();
- }
- } catch (Exception e) {
- LOG.warn("Error during stopping CamelContext. This exception is ignored.", e);
- } finally {
- // trigger stopping the Main
- latch.countDown();
- // shutdown the pool
- executorService.shutdownNow();
- }
+ // use thread to stop Camel as otherwise we would block current thread
+ Thread thread = camelContext.getExecutorServiceManager().newThread("CamelMainShutdownCamelContext", new ShutdownTask());
+ thread.start();
}
}
};
- executorService.scheduleAtFixedRate(task, 1, 1, TimeUnit.SECONDS);
+ idleExecutorService.scheduleAtFixedRate(task, 1, 1, TimeUnit.SECONDS);
+ }
+ }
+
+ private class ShutdownTask implements Runnable {
+
+ @Override
+ public void run() {
+ try {
+ // shutting down CamelContext
+ if (stopCamelContext) {
+ camelContext.stop();
+ }
+ } catch (Exception e) {
+ LOG.warn("Error during stopping CamelContext. This exception is ignored.", e);
+ } finally {
+ // trigger stopping the Main
+ latch.countDown();
+ }
}
}
diff --git a/core/camel-core/src/main/java/org/apache/camel/main/MainSupport.java b/core/camel-core/src/main/java/org/apache/camel/main/MainSupport.java
index a7869dd..0f3570e 100644
--- a/core/camel-core/src/main/java/org/apache/camel/main/MainSupport.java
+++ b/core/camel-core/src/main/java/org/apache/camel/main/MainSupport.java
@@ -639,18 +639,21 @@ public abstract class MainSupport extends ServiceSupport {
protected void waitUntilCompleted() {
while (!completed.get()) {
try {
+ int idle = mainConfigurationProperties.getDurationMaxIdleSeconds();
+ int max = mainConfigurationProperties.getDurationMaxMessages();
if (mainConfigurationProperties.getDuration() > 0) {
LOG.info("Waiting for: {} seconds", mainConfigurationProperties.getDuration());
latch.await(mainConfigurationProperties.getDuration(), TimeUnit.SECONDS);
exitCode.compareAndSet(UNINITIALIZED_EXIT_CODE, mainConfigurationProperties.getDurationHitExitCode());
completed.set(true);
- } else if (mainConfigurationProperties.getDurationMaxIdleSeconds() > 0) {
- LOG.info("Waiting to be idle for: {} seconds", mainConfigurationProperties.getDurationMaxIdleSeconds());
- exitCode.compareAndSet(UNINITIALIZED_EXIT_CODE, mainConfigurationProperties.getDurationHitExitCode());
- latch.await();
- completed.set(true);
- } else if (mainConfigurationProperties.getDurationMaxMessages() > 0) {
- LOG.info("Waiting until: {} messages has been processed", mainConfigurationProperties.getDurationMaxMessages());
+ } else if (idle > 0 || max > 0) {
+ if (idle > 0 && max > 0) {
+ LOG.info("Waiting to be idle for: {} seconds or until: {} messages has been processed", idle, max);
+ } else if (idle > 0) {
+ LOG.info("Waiting to be idle for: {} seconds", idle);
+ } else {
+ LOG.info("Waiting until: {} messages has been processed", max);
+ }
exitCode.compareAndSet(UNINITIALIZED_EXIT_CODE, mainConfigurationProperties.getDurationHitExitCode());
latch.await();
completed.set(true);
@@ -658,6 +661,9 @@ public abstract class MainSupport extends ServiceSupport {
latch.await();
}
} catch (InterruptedException e) {
+ // okay something interrupted us so terminate
+ completed.set(true);
+ latch.countDown();
Thread.currentThread().interrupt();
}
}