From 6b8bdc7e5e69b58282a456d5cc46e8cf252512d6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 5 Aug 2014 22:17:50 +0300 Subject: [PATCH] INT-3455 Orderly Shutdown Improvements JIRA: https://jira.spring.io/browse/INT-3455 Do not stop schedulers and executors - allows mid-flow QueueChannels to be drained. Stop all inbound MessageProducers (that are not OrderlyShutdownCapable). INT-3455 Polishing; PR Comments Change deprecate method usage to the new version --- .../jmx/config/MBeanExporterParser.java | 1 - .../monitor/IntegrationMBeanExporter.java | 238 +++++------------- .../jmx/config/spring-integration-jmx-4.1.xsd | 19 -- .../MBeanExporterIntegrationTests.java | 68 ++++- .../integration/monitor/lifecycle-source.xml | 16 +- .../monitor/self-destruction-context.xml | 8 +- src/reference/docbook/shutdown.xml | 35 +-- src/reference/docbook/whats-new.xml | 7 + 8 files changed, 162 insertions(+), 230 deletions(-) diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java index 409a1ad6f6..bf4b665b2d 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java @@ -58,7 +58,6 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-domain"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-name-static-properties"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "managed-components", "componentNamePatterns"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "shutdown-executor"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-naming-strategy", "namingStrategy"); builder.addPropertyValue("server", mbeanServer); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index fdb15eb68c..088efe1b5a 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 the original author or authors. * * Licensed 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 @@ -14,17 +14,12 @@ package org.springframework.integration.monitor; import java.lang.reflect.Field; -import java.util.ArrayList; import java.util.HashMap; import java.util.HashSet; -import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Properties; import java.util.Set; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; @@ -56,12 +51,13 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; -import org.springframework.core.task.support.ExecutorServiceAdapter; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.history.MessageHistoryConfigurer; import org.springframework.integration.support.context.NamedComponent; import org.springframework.jmx.export.MBeanExporter; @@ -76,10 +72,7 @@ import org.springframework.jmx.export.naming.MetadataNamingStrategy; import org.springframework.jmx.support.MetricType; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.PatternMatchUtils; import org.springframework.util.ReflectionUtils; @@ -115,7 +108,7 @@ import org.springframework.util.ReflectionUtils.FieldFilter; */ @ManagedResource public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware, - ApplicationContextAware, BeanClassLoaderAware, SmartLifecycle, Runnable { + ApplicationContextAware, BeanClassLoaderAware, SmartLifecycle { private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class); @@ -135,6 +128,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private final Set sources = new HashSet(); + private final Set inboundLifecycleMessageProducers = new HashSet(); + private final Set channels = new HashSet(); private final Map exposedBeans = new HashMap(); @@ -173,12 +168,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private String[] componentNamePatterns = { "*" }; - private volatile Executor shutdownExecutor; - private volatile long shutdownDeadline; - private volatile boolean shutdownForced; - private final AtomicBoolean shuttingDown = new AtomicBoolean(); private MessageHistoryConfigurer messageHistoryConfigurer; @@ -236,11 +227,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP this.applicationContext = applicationContext; } - public void setShutdownExecutor(Executor shutdownExecutor) { - Assert.notNull(shutdownExecutor, "Shutdown Executor may not be null"); - this.shutdownExecutor = shutdownExecutor; - } - @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { @@ -300,6 +286,13 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP bean = advised; } + if (bean instanceof MessageProducer && bean instanceof Lifecycle) { + Lifecycle target = (Lifecycle) extractTarget(bean); + if (!(target instanceof AbstractReplyProducingMessageHandler)) { // TODO: change to AMPMH + this.inboundLifecycleMessageProducers.add(target); + } + } + return bean; } @@ -478,89 +471,66 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } /** - * Shutdown active components. If the thread calling this method is - * managed by a Spring-managed executor, you should provide a specific - * dedicated executor via the {@link #setShutdownExecutor} - * method. When this is provided, the shutdown will be performed on one - * of its threads, instead of the calling thread; thus avoiding - * the situation where we will wait for the current thread to terminate. - *

It is not necessary to supply this executor service if the - * current thread will not, itself, be shutdown as a result of - * calling this method. - *

Note: The supplied executor service - * will not be shut down. + * Shutdown active components. + * + * @param force No longer used. + * @param howLong The time to wait in total for all activities to complete + * in milliseconds. + * @deprecated Use {@link #stopActiveComponents(long)}. + */ + @Deprecated + @ManagedOperation + public void stopActiveComponents(boolean force, long howLong) { + stopActiveComponents(howLong); + } + + /** + * Shutdown active components. * - * @param force If true, stop the executors with shutdownNow(), canceling - * running tasks. Overrides any settings on schedulers/executors. When true - * may result in error messages being sent to error channels. * @param howLong The time to wait in total for all activities to complete * in milliseconds. */ @ManagedOperation - public void stopActiveComponents(boolean force, long howLong) { + public void stopActiveComponents(long howLong) { if (!this.shuttingDown.compareAndSet(false, true)) { logger.error("Shutdown already in process"); return; } this.shutdownDeadline = System.currentTimeMillis() + howLong; - this.shutdownForced = force; - if (this.shutdownExecutor == null) { - try { - logger.debug("Running shutdown on current thread"); - this.run(); - } catch (Exception e) { - logger.error("Orderly shutdown failed", e); - } + try { + logger.debug("Running shutdown"); + doShutdown(); } - else { - logger.debug("Launching shutdown on another thread"); - this.shutdownExecutor.execute(this); + catch (Exception e) { + logger.error("Orderly shutdown failed", e); } } /** * Perform orderly shutdown - called or executed from - * {@link #stopActiveComponents(boolean, long)}. + * {@link #stopActiveComponents(long)}. */ - @Override - public void run() { + private void doShutdown() { try { - this.orderlyShutdownCapableComponentsBefore(); - this.stopActiveChannels(); - this.stopSchedulers(); - if (System.currentTimeMillis() > this.shutdownDeadline) { - logger.error("Timed out before waiting for all schedulers to complete"); - } - this.stopExecutors(); - if (System.currentTimeMillis() > this.shutdownDeadline) { - logger.error("Timed out before waiting for all executors to complete"); - } - this.stopNonSpringExecutors(); - if (System.currentTimeMillis() > this.shutdownDeadline) { - logger.error("Timed out before waiting for all non-Spring executors to complete"); - } - this.stopMessageSources(); + orderlyShutdownCapableComponentsBefore(); + stopActiveChannels(); + stopMessageSources(); + stopInboundMessageProducers(); // Wait any remaining time for messages to quiesce - long timeLeft = this.shutdownDeadline - System.currentTimeMillis(); + long timeLeft = shutdownDeadline - System.currentTimeMillis(); if (timeLeft > 0) { try { Thread.sleep(timeLeft); - } catch (InterruptedException e) { + } + catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.error("Interrupted while waiting for quiesce"); } - this.orderlyShutdownCapableComponentsAfter(); - } - else { - this.shutdownForced = true; - this.stopSchedulers(); - this.stopExecutors(); - this.stopNonSpringExecutors(); - this.orderlyShutdownCapableComponentsAfter(); } + orderlyShutdownCapableComponentsAfter(); } finally { - this.shuttingDown.set(false); + shuttingDown.set(false); } } @@ -585,6 +555,22 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + /** + * Stops all inbound message producers (that are not {@link OrderlyShutdownCapable}) + * - may cause interrupts. + */ + @ManagedOperation + public void stopInboundMessageProducers() { + for (Lifecycle producer : this.inboundLifecycleMessageProducers) { + if (!(producer instanceof OrderlyShutdownCapable)) { + if (logger.isInfoEnabled()) { + logger.info("Stopping message producer " + producer); + } + producer.stop(); + } + } + } + @ManagedOperation public void stopActiveChannels() { // Stop any "active" channels (JMS etc). @@ -600,80 +586,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } - @ManagedOperation - public void stopSchedulers() { - if (logger.isDebugEnabled()) { - logger.debug("Stopping schedulers " + (this.shutdownForced ? "(force)" : "")); - } - List executorServices = new ArrayList(); - Map schedulers = this.applicationContext - .getBeansOfType(ThreadPoolTaskScheduler.class); - for (Entry entry : schedulers.entrySet()) { - ThreadPoolTaskScheduler scheduler = entry.getValue(); - if (logger.isInfoEnabled()) { - logger.info("Stopping scheduler " + scheduler.getThreadNamePrefix()); - } - ExecutorService executorService = scheduler.getScheduledExecutor(); - executorServices.add(executorService); - doShutdownExecutorService(executorService); - } - waitForExecutors(executorServices); - logger.debug("Stopped schedulers"); - } - - @ManagedOperation - public void stopExecutors() { - if (logger.isDebugEnabled()) { - logger.debug("Stopping executors" + (this.shutdownForced ? "(force)" : "")); - } - List executorServices = new ArrayList(); - Map executors = this.applicationContext - .getBeansOfType(ThreadPoolTaskExecutor.class); - for (Entry entry : executors.entrySet()) { - ThreadPoolTaskExecutor executor = entry.getValue(); - if (executor == this.shutdownExecutor) { - logger.debug("Skipping shutdown of shutdown executor"); - } - else { - if (logger.isInfoEnabled()) { - logger.info("Stopping executor " + executor.getThreadNamePrefix()); - } - ExecutorService executorService = executor.getThreadPoolExecutor(); - executorServices.add(executorService); - doShutdownExecutorService(executorService); - } - } - waitForExecutors(executorServices); - logger.debug("Stopped executors"); - } - - @ManagedOperation - public void stopNonSpringExecutors() { - if (logger.isDebugEnabled()) { - logger.debug("Stopping other executors" + (this.shutdownForced ? "(force)" : "")); - } - List executorServices = new ArrayList(); - Map nonSpringExecutors = this.applicationContext - .getBeansOfType(ExecutorService.class); - for (Entry entry : nonSpringExecutors.entrySet()) { - ExecutorService executorService = entry.getValue(); - if (!(executorService instanceof ExecutorServiceAdapter)) { - if (logger.isInfoEnabled()) { - logger.info("Stopping executor service " + executorService); - } - executorServices.add(executorService); - doShutdownExecutorService(executorService); - } - else { - if (logger.isDebugEnabled()) { - logger.debug("Ignoring ExecutorServiceAdapter"); - } - } - } - waitForExecutors(executorServices); - logger.debug("Stopped other executors"); - } - protected final void orderlyShutdownCapableComponentsBefore() { logger.debug("Initiating stop OrderlyShutdownCapable components"); Map components = this.applicationContext @@ -793,33 +705,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } - private void doShutdownExecutorService(ExecutorService executorService) { - if (this.shutdownForced) { - executorService.shutdownNow(); - } - else { - executorService.shutdown(); - } - } - - private void waitForExecutors(List executorServices) { - for (ExecutorService executorService : executorServices) { - try { - if (!executorService.awaitTermination(this.shutdownDeadline - - System.currentTimeMillis(), TimeUnit.MILLISECONDS)) { - logger.error("Executor service " + executorService + " failed to terminate"); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.error("Interrupted while shutting down executor service " + executorService); - throw new MessagingException("Interrupted while shutting down", e); - } - if (System.currentTimeMillis() > this.shutdownDeadline) { - logger.error("Timed out before waiting for all executor services"); - } - } - } - private void registerChannels() { for (DirectChannelMetrics monitor : channels) { String name = monitor.getName(); @@ -1231,4 +1116,5 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP ReflectionUtils.makeAccessible(field); return ReflectionUtils.getField(field, target); } + } diff --git a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.1.xsd b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.1.xsd index 06f3626832..7c0d5ff9ec 100644 --- a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.1.xsd +++ b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.1.xsd @@ -186,25 +186,6 @@ - - - - - - - - - An Executor used when shutting down the application using the 'stopActiveComponents()' - method. Only required when invoking the operation on a Spring-managed thread, such as - via a from, say, an error flow. Using this executor avoids the - problem where the shutdown will wait for the current thread to terminate, time out, - and then force-close other components. When a dedicated executor is supplied, - the method will not wait on its threads to complete, and will terminate normally. - It is recommended that the executor used here is dedicated for this purpose and - not used elsewhere. - - - diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java index ade50b1655..32cfa7b55c 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2012 the original author or authors. + * Copyright 2009-2014 the original author or authors. * * Licensed 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 @@ -28,17 +28,22 @@ import javax.management.ObjectName; import org.junit.After; import org.junit.Test; + +import org.springframework.aop.framework.Advised; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.context.support.GenericXmlApplicationContext; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.util.Assert; public class MBeanExporterIntegrationTests { @@ -100,7 +105,7 @@ public class MBeanExporterIntegrationTests { assertNotNull(messageChannelsMonitor); MBeanServer server = context.getBean(MBeanServer.class); Set names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null); - assertEquals(0, names.size()); + assertEquals(2, names.size()); names = server.queryNames(ObjectName.getInstance("org.springframework.integration:name=explicit,*"), null); assertEquals(1, names.size()); MBeanOperationInfo[] operations = server.getMBeanInfo(names.iterator().next()).getOperations(); @@ -114,13 +119,39 @@ public class MBeanExporterIntegrationTests { // Lifecycle method name assertEquals("start", startName); assertTrue((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null)); - messageChannelsMonitor.stopActiveComponents(false, 3000); + messageChannelsMonitor.stopActiveComponents(3000); assertFalse((Boolean) server.invoke(names.iterator().next(), "isRunning", null, null)); ActiveChannel activeChannel = context.getBean("activeChannel", ActiveChannel.class); assertTrue(activeChannel.isStopCalled()); OtherActiveComponent otherActiveComponent = context.getBean(OtherActiveComponent.class); assertTrue(otherActiveComponent.isBeforeCalled()); assertTrue(otherActiveComponent.isAfterCalled()); + assertTrue(otherActiveComponent.isRunning()); + assertFalse(context.getBean(AMessageProducer.class).isRunning()); + + // check pollers are still running + QueueChannel input = (QueueChannel) extractTarget(context.getBean("input")); + QueueChannel input2 = (QueueChannel) extractTarget(context.getBean("input2")); + input.purge(null); + input2.purge(null); + input.send(new GenericMessage("foo")); + assertNotNull(input2.receive(10000)); + } + + private Object extractTarget(Object bean) { + if (!(bean instanceof Advised)) { + return bean; + } + Advised advised = (Advised) bean; + if (advised.getTargetSource() == null) { + return null; + } + try { + return extractTarget(advised.getTargetSource().getTarget()); + } + catch (Exception e) { + return null; + } } @Test @@ -214,15 +245,18 @@ public class MBeanExporterIntegrationTests { // This has the potential to blow up if called before setDate(). // Depends on bean instantiation order and IntegrationMBeanExporter // can influence that by aggressively instantiating other MBeanExporters + @Override public Date getObject() throws Exception { Assert.state(date != null, "A date must be provided"); return date; } + @Override public Class getObjectType() { return Date.class; } + @Override public boolean isSingleton() { return true; } @@ -237,6 +271,7 @@ public class MBeanExporterIntegrationTests { this.date = date; } + @Override public void afterPropertiesSet() throws Exception { Assert.state(date != null, "A date must be provided"); } @@ -246,21 +281,24 @@ public class MBeanExporterIntegrationTests { public static class MetricFactoryBean implements FactoryBean { private MessageChannel channel; - private Metric date = new Metric(); + private final Metric date = new Metric(); public void setChannel(MessageChannel channel) { this.channel = channel; } + @Override public Metric getObject() throws Exception { Assert.state(channel != null, "A channel must be provided"); return date; } + @Override public Class getObjectType() { return Metric.class; } + @Override public boolean isSingleton() { return true; } @@ -280,6 +318,7 @@ public class MBeanExporterIntegrationTests { this.channel = channel; } + @Override public void afterPropertiesSet() throws Exception { Assert.state(channel != null, "A channel must be provided"); } @@ -294,12 +333,14 @@ public class MBeanExporterIntegrationTests { public static class SimpleService implements Service { private int counter; + @Override public String execute() throws Exception { Thread.sleep(10L); // make the duration non-zero counter++; return "count="+counter; } + @Override public int getCounter() { return counter; } @@ -313,31 +354,38 @@ public class MBeanExporterIntegrationTests { private boolean stopCalled; + @Override public boolean send(Message message) { return false; } + @Override public boolean send(Message message, long timeout) { return false; } + @Override public void start() { } + @Override public void stop() { this.stopCalled = true; } + @Override public boolean isRunning() { return false; } + @Override public boolean isStopCalled() { return this.stopCalled; } } - public static class OtherActiveComponent implements OrderlyShutdownCapable { + public static class OtherActiveComponent extends MessageProducerSupport + implements OrderlyShutdownCapable { private boolean beforeCalled; @@ -351,14 +399,20 @@ public class MBeanExporterIntegrationTests { return afterCalled; } + @Override public int beforeShutdown() { this.beforeCalled = true; return 0; } + @Override public int afterShutdown() { this.afterCalled = true; return 0; } } + + public static class AMessageProducer extends MessageProducerSupport { + } + } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml index 4f6bb43549..4054bbd3d0 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml @@ -15,6 +15,14 @@ + + + + + + + + @@ -29,7 +37,13 @@ - + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml index 99962e0271..31719521bb 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/self-destruction-context.xml @@ -18,15 +18,13 @@ - - - @@ -35,6 +33,8 @@ - + + + diff --git a/src/reference/docbook/shutdown.xml b/src/reference/docbook/shutdown.xml index 21c76edddb..683c3417dd 100644 --- a/src/reference/docbook/shutdown.xml +++ b/src/reference/docbook/shutdown.xml @@ -11,9 +11,8 @@ As described in , the MBean exporter provides a JMX operation stopActiveComponents, which is used to stop the application in an orderly manner. The operation - has two parameters, a boolean and a long. The boolean indicates whether attempts will be made - to stop (interrupt) active threads; in most cases this will be set to false for orderly - shutdown. The long parameter indicates how long (in milliseconds) the operation will wait to allow + has a single long parameter. + The parameter indicates how long (in milliseconds) the operation will wait to allow in-flight messages to complete. The operation works as follows: @@ -29,35 +28,21 @@ The second step stops any active channels, such as JMS- or AMQP-backed channels. - The third step stops all TaskSchedulers, preventing any new - scheduled operations (polling etc). + The third step stops all MessageSources. - The fourth step stops all TaskExecutors, preventing any new - tasks from running. - - - If the shutdown is running from a Spring-managed TaskExecutor, shutting down that - executor would cause all the timeout time to be consumed by this step, because the thread won't terminate). - For this reason, either use a dedicated executor (via the shutdownExecutor property on the MBean exporter), - or do not use a Spring-managed executor to invoke this operation. - - - The fifth step stops all MessageSources. + The fourth step stops all inbound MessageProducers (that are not + OrderlyShutdownCapable). - The sixth step waits for any remaining time left, as defined by the value of the long parameter passed + The fifth step waits for any remaining time left, as defined by the value of the long parameter passed in to the operation. This is intended to allow any in-flight messages to complete their journeys. It is therefore important to select an appropriate timeout when invoking this operation. - The seventh step calls afterShutdown() on all OrderlyShutdownCapable components. + The sixth step calls afterShutdown() on all OrderlyShutdownCapable components. This allows such components to perform final shutdown tasks (closing all open sockets, for example). - - If no time is left when we get to step 6, it probably means some thread is hung; in which case, the - operation attempts a forced shutdown on all schedulers and executors before exiting. - As discussed in this operation can be invoked using JMX. If you wish to programmatically invoke the method, you will need to inject, or otherwise get a reference to, @@ -76,4 +61,10 @@ monitoring Spring Integration sample application for details. + + The above algorithm was improved in version 4.1. Previously, all task executors + and schedulers were stopped. This could cause mid-flow messages in QueueChannels + to remain. Now, the shutdown leaves pollers running in order to allow these messages to be + drained and processed. + diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 8b44fb1568..6ed8e5a15c 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -119,5 +119,12 @@ See for more information. +

+ Orderly Shutdown + + Improvements have been made to the orderly shutdown algorithm. + See for more information. + +