INT-1132: Rescheduling Support for DelayHandler

- Add support for DelayHandler to reschedule persisted Messages on startup via 'implements ApplicationListener<ContextRefreshedEvent>' as late as possible
- Change DelayHandler dependency to MessageGroupStore
- Add required DelayHandler.messageGroupId property
- Make registered by SI-namespace TaskScheduler as default for DelayHandler
- Remove 'required' from delayer xml-attribute 'default-delay' as redundant
- Additional refactoring & polishing around <delayer>
- Polishing delayer's Tests
- Tests for 'rescheduling'
- Integration test for 'rescheduling' with JdbcMS

INT-1132: add 'initializingLatch' to DelayHandler

INT-1132: additional polishing

INT-1132: add LIFO JMS polling test-case

INT-1132: Changes according to PR comments

INT-1132: Changes according to PR comments 2

Polishing

Remove blocking calls from tests.

spring-integration-core/src/main/java/org/springframework/integration/config/xml/DelayerParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.config.xml;
 
+import org.springframework.integration.handler.DelayHandler;
 import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
@@ -24,30 +25,49 @@ import org.springframework.util.StringUtils;
 
 /**
  * Parser for the <delayer> element.
- * 
+ *
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 1.0.3
  */
 public class DelayerParser extends AbstractConsumerEndpointParser {
 
 	@Override
 	protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
-		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
-				IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.DelayHandler");
-		String defaultDelay = element.getAttribute("default-delay");
-		if (!StringUtils.hasText(defaultDelay)) {
-			parserContext.getReaderContext().error("The 'default-delay' attribute is required.", element);
-			return null;
+		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(DelayHandler.class);
+
+		String id = element.getAttribute(ID_ATTRIBUTE);
+		if (!StringUtils.hasText(id)) {
+			parserContext.getReaderContext().error("The 'id' attribute is required.", element);
 		}
-		builder.addConstructorArgValue(defaultDelay);
+
+		String defaultDelay = element.getAttribute("default-delay");
+		String delayHeaderName = element.getAttribute("delay-header-name");
+
+		boolean hasDefaultDelay = StringUtils.hasText(defaultDelay);
+		boolean hasDelayHeaderName = StringUtils.hasText(delayHeaderName);
+
+		if (!(hasDefaultDelay | hasDelayHeaderName)) {
+			parserContext.getReaderContext()
+					.error("The 'default-delay' or 'delay-header-name' attributes should be provided.", element);
+		}
+
+		builder.addConstructorArgValue(id + ".messageGroupId");
+
 		String scheduler = element.getAttribute("scheduler");
 		if (StringUtils.hasText(scheduler)) {
 			builder.addConstructorArgReference(scheduler);
 		}
+
+		if (hasDefaultDelay) {
+			builder.addPropertyValue("defaultDelay", defaultDelay);
+		}
+		if (hasDelayHeaderName) {
+			builder.addPropertyValue("delayHeaderName", delayHeaderName);
+		}
+
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-store");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "delay-header-name");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "wait-for-tasks-to-complete-on-shutdown");
 		return builder;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2011 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -16,31 +16,21 @@
 
 package org.springframework.integration.handler;
 
+import java.io.Serializable;
 import java.util.Date;
-import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.core.Ordered;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.integration.Message;
-import org.springframework.integration.MessageChannel;
-import org.springframework.integration.MessageHandlingException;
-import org.springframework.integration.MessageHeaders;
-import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.MessageHandler;
-import org.springframework.integration.core.MessageProducer;
-import org.springframework.integration.core.MessagingTemplate;
-import org.springframework.integration.message.ErrorMessage;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.store.MessageStore;
 import org.springframework.integration.store.SimpleMessageStore;
-import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
-import org.springframework.integration.support.channel.ChannelResolutionException;
-import org.springframework.integration.support.channel.ChannelResolver;
+import org.springframework.integration.support.MessageBuilder;
 import org.springframework.scheduling.TaskScheduler;
-import org.springframework.scheduling.concurrent.ExecutorConfigurationSupport;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.Assert;
 
@@ -52,12 +42,12 @@ import org.springframework.util.Assert;
  * therefore, the calling thread does not block. The advantage of this approach
  * is that many delays can be managed concurrently, even very long delays,
  * without producing a buildup of blocked Threads.
- * <p>
+ * <p/>
  * One thing to keep in mind, however, is that any active transactional context
  * will not propagate from the original sender to the eventual recipient. This
  * is a side-effect of passing the Message to the output channel after the
  * delay with a different Thread in control.
- * <p>
+ * <p/>
  * When this handler's 'delayHeaderName' property is configured, that value, if
  * present on a Message, will take precedence over the handler's 'defaultDelay'
  * value. The actual header value may be a long, a String that can be parsed
@@ -67,54 +57,49 @@ import org.springframework.util.Assert;
  * seconds from the current time). If the value is a Date, it will be
  * delayed at least until that Date occurs (i.e. the delay in that case is
  * equivalent to <code>headerDate.getTime() - new Date().getTime()</code>).
- * 
+ *
  * @author Mark Fisher
+ * @author Artem Bilan
  * @since 1.0.3
  */
-public class DelayHandler extends IntegrationObjectSupport implements MessageHandler, MessageProducer, Ordered, DisposableBean {
+public class DelayHandler extends AbstractReplyProducingMessageHandler implements ApplicationListener<ContextRefreshedEvent> {
 
-	private final Log logger = LogFactory.getLog(this.getClass());
+	private final String messageGroupId;
 
 	private volatile long defaultDelay;
 
 	private volatile String delayHeaderName;
 
-	private boolean waitForTasksToCompleteOnShutdown = false;
-
-	private volatile MessageChannel outputChannel;
-
-	private volatile ChannelResolver channelResolver;
-
-	private volatile MessageStore messageStore;
-
-	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
-
-	private volatile int order = Ordered.LOWEST_PRECEDENCE;
+	private volatile MessageGroupStore messageStore;
 
+	private final AtomicBoolean initialized = new AtomicBoolean();
 
 	/**
-	 * Create a DelayHandler with the given default delay. The sending of Messages after
-	 * the delay will be handled by a scheduled thread pool with a size of 1.
+	 * Create a DelayHandler with the given 'messageGroupId' that is used as 'key' for {@link MessageGroup}
+	 * to store delayed Messages in the {@link MessageGroupStore}. The sending of Messages after
+	 * the delay will be handled by registered in the ApplicationContext default {@link ThreadPoolTaskScheduler}.
+	 *
+	 * @see IntegrationObjectSupport#getTaskScheduler()
 	 */
-	public DelayHandler(long defaultDelay) {
-		this(defaultDelay, null);
+	public DelayHandler(String messageGroupId) {
+		Assert.notNull(messageGroupId, "'messageGroupId' must not be null");
+		this.messageGroupId = messageGroupId;
 	}
 
 	/**
 	 * Create a DelayHandler with the given default delay. The sending of Messages
 	 * after the delay will be handled by the provided {@link TaskScheduler}.
 	 */
-	public DelayHandler(long defaultDelay, TaskScheduler taskScheduler) {
-		this.defaultDelay = defaultDelay;
-		this.setTaskScheduler(taskScheduler != null ? taskScheduler : new ThreadPoolTaskScheduler());
+	public DelayHandler(String messageGroupId, TaskScheduler taskScheduler) {
+		this(messageGroupId);
+		this.setTaskScheduler(taskScheduler);
 	}
 
-
 	/**
 	 * Set the default delay in milliseconds. If no 'delayHeaderName' property
 	 * has been provided, the default delay will be applied to all Messages. If
 	 * a delay should <emphasis>only</emphasis> be applied to Messages with a
-	 * header, then set this value to 0. 
+	 * header, then set this value to 0.
 	 */
 	public void setDefaultDelay(long defaultDelay) {
 		this.defaultDelay = defaultDelay;
@@ -130,82 +115,57 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan
 	}
 
 	/**
-	 * Specify the {@link MessageStore} that should be used to store Messages
+	 * Specify the {@link MessageGroupStore} that should be used to store Messages
 	 * while awaiting the delay.
 	 */
-	public void setMessageStore(MessageStore messageStore) {
+	public void setMessageStore(MessageGroupStore messageStore) {
+		Assert.state(messageStore != null, "MessageStore must not be null");
 		this.messageStore = messageStore;
 	}
 
-	/**
-	 * Set the output channel for this handler. If none is provided, each
-	 * inbound Message must include a reply channel header.
-	 */
-	public void setOutputChannel(MessageChannel outputChannel) {
-		this.outputChannel = outputChannel;
-	}
-
-	/**
-	 * Set the timeout for sending reply Messages.
-	 */
-	public void setSendTimeout(long sendTimeout) {
-		this.messagingTemplate.setSendTimeout(sendTimeout);
-	}
-
-	/**
-	 * Set whether to wait for scheduled tasks to complete on shutdown.
-	 * <p>Default is "false". Switch this to "true" if you prefer
-	 * fully completed tasks at the expense of a longer shutdown phase.
-	 * <p>
-	 * This property will only have an effect for TaskScheduler implementations
-	 * that extend from {@link ExecutorConfigurationSupport}.
-	 * @see ExecutorConfigurationSupport#setWaitForTasksToCompleteOnShutdown(boolean)
-	 */
-	public void setWaitForTasksToCompleteOnShutdown(boolean waitForJobsToCompleteOnShutdown) {
-		this.waitForTasksToCompleteOnShutdown = waitForJobsToCompleteOnShutdown;
-	}
-
-	public void setOrder(int order) {
-		this.order = order;
-	}
-
-	public int getOrder() {
-		return this.order;
-	}
-
 	@Override
 	public String getComponentType() {
 		return "delayer";
 	}
 
-	protected void onInit() throws Exception{
-		if (this.getTaskScheduler() instanceof ExecutorConfigurationSupport) {
-			((ExecutorConfigurationSupport) this.getTaskScheduler()).setWaitForTasksToCompleteOnShutdown(this.waitForTasksToCompleteOnShutdown);
-		}
-		else if (logger.isWarnEnabled()) {
-			logger.warn("The 'waitForJobsToCompleteOnShutdown' property is not supported for TaskScheduler of type [" +
-					this.getTaskScheduler().getClass() + "]");
-		}
+	@Override
+	protected void onInit() {
+		super.onInit();
 		if (this.messageStore == null) {
 			this.messageStore = new SimpleMessageStore();
 		}
-		if (this.getTaskScheduler() instanceof InitializingBean) {
-			((InitializingBean) this.getTaskScheduler()).afterPropertiesSet();
-		}
-		if (this.getBeanFactory() != null){
-			this.channelResolver = new BeanFactoryChannelResolver(this.getBeanFactory());
+		else {
+			Assert.isInstanceOf(MessageStore.class, this.messageStore);
 		}
 	}
 
-	public final void handleMessage(final Message<?> message) {
-		long delay = this.determineDelayForMessage(message);
-		if (delay > 0) {
-			this.releaseMessageAfterDelay(message, delay);
-		}
-		else {
-			// no delay, send directly
-			this.sendMessageToReplyChannel(message);
+	/**
+	 * Checks if 'requestMessage' wasn't delayed before
+	 * ({@link #releaseMessageAfterDelay} and {@link DelayedMessageWrapper}).
+	 * Than determine 'delay' for 'requestMessage' ({@link #determineDelayForMessage})
+	 * and if <code>delay > 0</code> schedules 'releaseMessage' task after 'delay'.
+	 *
+	 * @param requestMessage - the Message which may be delayed.
+	 * @return - <code>null</code> if 'requestMessage' is delayed,
+	 * 			 otherwise - 'payload' from 'requestMessage'.
+	 * @see #releaseMessage
+	 */
+
+	@Override
+	protected Object handleRequestMessage(Message<?> requestMessage) {
+		boolean delayed = requestMessage.getPayload() instanceof DelayedMessageWrapper;
+
+		if (!delayed) {
+			long delay = this.determineDelayForMessage(requestMessage);
+			if (delay > 0) {
+				this.releaseMessageAfterDelay(requestMessage, delay);
+				return null;
+			}
 		}
+
+		// no delay
+		Object payload = requestMessage.getPayload();
+		return delayed ? ((DelayedMessageWrapper) payload).getOriginal().getPayload() : payload;
 	}
 
 	private long determineDelayForMessage(Message<?> message) {
@@ -231,98 +191,116 @@ public class DelayHandler extends IntegrationObjectSupport implements MessageHan
 	}
 
 	private void releaseMessageAfterDelay(final Message<?> message, long delay) {
-		Assert.state(this.messageStore != null, "MessageStore must not be null");
-		final Message<?> storedMessage = this.messageStore.addMessage(message);
+		Message<?> delayedMessage = message;
+
+		DelayedMessageWrapper messageWrapper = null;
+		if (message.getPayload() instanceof DelayedMessageWrapper) {
+			messageWrapper = (DelayedMessageWrapper) message.getPayload();
+		}
+		else {
+			messageWrapper = new DelayedMessageWrapper(message);
+			delayedMessage = MessageBuilder.withPayload(messageWrapper).copyHeaders(message.getHeaders()).build();
+			this.messageStore.addMessageToGroup(this.messageGroupId, delayedMessage);
+		}
+
+		final Message<?> messageToSchedule = delayedMessage;
+
 		this.getTaskScheduler().schedule(new Runnable() {
 			public void run() {
-				try {
-					releaseMessage(storedMessage.getHeaders().getId());
-				}
-				catch (Exception e) {
-					Exception exception = new MessageHandlingException(message, "Failed to deliver Message after delay.", e);
-					MessageChannel errorChannel = resolveErrorChannelIfPossible(message);
-					if (errorChannel != null) {
-						ErrorMessage errorMessage = new ErrorMessage(exception);
-						try {
-							messagingTemplate.send(errorChannel, errorMessage);
-						}
-						catch (Exception e2) {
-							if (logger.isWarnEnabled()) {
-								logger.warn("Failed to send MessagingException to error channel.", exception);
-							}
-						}
+				releaseMessage(messageToSchedule);
+			}
+		}, new Date(messageWrapper.getRequestDate() + delay));
+	}
+
+	private void releaseMessage(Message<?> message) {
+		if (this.messageStore instanceof SimpleMessageStore
+				|| ((MessageStore) this.messageStore).removeMessage(message.getHeaders().getId()) != null) {
+			this.messageStore.removeMessageFromGroup(this.messageGroupId, message);
+			this.handleMessageInternal(message);
+		}
+		else {
+			if (logger.isDebugEnabled()) {
+				logger.debug("No message in the Message Store to release: " + message +
+						". Likely another instance has already released it.");
+			}
+		}
+	}
+
+	/**
+	 * Used for reading persisted Messages in the 'messageStore'
+	 * to reschedule them e.g. upon application restart.
+	 * The logic is based on iteration over 'messageGroup.getMessages()'
+	 * and schedules task about 'delay' logic.
+	 * This behavior is dictated by the avoidance of invocation thread overload.
+	 */
+	public void reschedulePersistedMessages() {
+		MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
+		for (final Message<?> message : messageGroup.getMessages()) {
+			this.getTaskScheduler().schedule(new Runnable() {
+				public void run() {
+					long delay = determineDelayForMessage(message);
+					if (delay > 0) {
+						releaseMessageAfterDelay(message, delay);
 					}
-					else if (logger.isWarnEnabled()) {
-						logger.warn("No error channel available. MessagingException will be ignored.", exception);
+					else {
+						releaseMessage(message);
 					}
 				}
-			}
-		}, new Date(System.currentTimeMillis() + delay));
+			}, new Date());
+		}
 	}
 
-	private void releaseMessage(UUID id) {
-		Assert.state(this.messageStore != null, "MessageStore must not be null");
-		Message<?> message = this.messageStore.removeMessage(id);
-		Assert.notNull(message, "Message with id: " + id + " no longer exists in MessageStore.");
-		this.sendMessageToReplyChannel(message);
+	/**
+	 * Handles {@link ContextRefreshedEvent} to invoke {@link #reschedulePersistedMessages}
+	 * as late as possible after application context startup.
+	 * Also it checks {@link #initialized} to ignore
+	 * other {@link ContextRefreshedEvent}s which may be published
+	 * in the 'parent-child' contexts, e.g. in the Spring-MVC applications.
+	 *
+	 * @param event - {@link ContextRefreshedEvent} which occurs
+	 *                 after Application context is completely initialized.
+	 * @see #reschedulePersistedMessages
+	 */
+	public void onApplicationEvent(ContextRefreshedEvent event) {
+		if (!this.initialized.getAndSet(true)) {
+			this.reschedulePersistedMessages();
+		}
 	}
 
-	private void sendMessageToReplyChannel(Message<?> message) {
-		MessageChannel replyChannel = this.resolveReplyChannel(message);
-		this.messagingTemplate.send(replyChannel, message);
+	private static final class DelayedMessageWrapper implements Serializable {
+
+		private static final long serialVersionUID = -4739802369074947045L;
+
+		private final long requestDate = System.currentTimeMillis();
+
+		private final Message<?> original;
+
+		public DelayedMessageWrapper(Message<?> original) {
+			this.original = original;
+		}
+
+		public long getRequestDate() {
+			return this.requestDate;
+		}
+
+		public Message<?> getOriginal() {
+			return this.original;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) return true;
+			if (o == null || getClass() != o.getClass()) return false;
+
+			DelayedMessageWrapper that = (DelayedMessageWrapper) o;
+
+			return this.original.equals(that.original);
+		}
+
+		@Override
+		public int hashCode() {
+			return this.original.hashCode();
+		}
 	}
 
-	private MessageChannel resolveReplyChannel(Message<?> message) {
-		MessageChannel replyChannel = this.outputChannel;
-		if (replyChannel == null) {
-			replyChannel = this.resolveChannelFromHeader(message, MessageHeaders.REPLY_CHANNEL);
-		}
-		if (replyChannel == null) {
-			throw new ChannelResolutionException(
-					"unable to resolve reply channel for message: " + message);
-		}
-		return replyChannel;
-	}
-
-	private MessageChannel resolveErrorChannelIfPossible(Message<?> message) {
-		MessageChannel errorChannel = null;
-		try {
-			errorChannel = this.resolveChannelFromHeader(message, MessageHeaders.ERROR_CHANNEL);
-		}
-		catch (Exception e) {
-			if (logger.isWarnEnabled()) {
-				logger.warn("Failed to resolve error channel from header.", e);
-			}
-		}
-		if (errorChannel == null && this.channelResolver != null) {
-			errorChannel = this.channelResolver.resolveChannelName(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
-		}
-		return errorChannel;
-	}
-
-	private MessageChannel resolveChannelFromHeader(Message<?> message, String headerName) {
-		MessageChannel channel = null;
-		Object channelHeader = message.getHeaders().get(headerName);
-		if (channelHeader != null) {
-			if (channelHeader instanceof MessageChannel) {
-				channel = (MessageChannel) channelHeader;
-			}
-			else if (channelHeader instanceof String) {
-				Assert.state(this.channelResolver != null,
-						"ChannelResolver is required for resolving '" + headerName + "' by name.");
-				channel = this.channelResolver.resolveChannelName((String) channelHeader);
-			}
-			else {
-				throw new ChannelResolutionException("expected a MessageChannel or String for '" +
-						headerName + "', but type is [" + channelHeader.getClass() + "]");
-			}
-		}
-		return channel;
-	}
-
-	public void destroy() throws Exception {
-		if (this.getTaskScheduler() instanceof DisposableBean) {
-			((DisposableBean) this.getTaskScheduler()).destroy();
-		}
-	}
 }

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd

@@ -716,7 +716,7 @@
 			</xsd:attribute>
 		</xsd:complexType>
 	</xsd:element>
-	
+
 	<xsd:element name="resource-inbound-channel-adapter">
 		<xsd:annotation>
 			<xsd:documentation>
@@ -770,11 +770,11 @@
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
-			
+
 			<xsd:attribute name="pattern-resolver" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>
-						Reference to a org.springframework.core.io.support.ResourcePatternResolver. 
+						Reference to a org.springframework.core.io.support.ResourcePatternResolver.
 					</xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>
@@ -986,7 +986,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 					<xsd:attribute name="requires-reply" type="xsd:string" use="optional">
 						<xsd:annotation>
 							<xsd:documentation>
@@ -1085,7 +1085,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="enricher-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1138,7 +1138,7 @@
         <xsd:attribute name="request-timeout" type="xsd:string">
             <xsd:annotation>
                 <xsd:documentation><![CDATA[
-                    Set the timeout value for sending request messages in milliseconds. 
+                    Set the timeout value for sending request messages in milliseconds.
                     If not explicitly configured, the default is one second.
                     ]]>
                 </xsd:documentation>
@@ -1147,7 +1147,7 @@
         <xsd:attribute name="reply-timeout" type="xsd:string">
             <xsd:annotation>
                 <xsd:documentation><![CDATA[
-                    Set the timeout value for receiving reply messages in milliseconds. 
+                    Set the timeout value for receiving reply messages in milliseconds.
                     If not explicitly configured, the default is one second.
                     ]]>
                 </xsd:documentation>
@@ -1156,25 +1156,25 @@
         <xsd:attribute name="requires-reply" use="optional" default="true">
         <xsd:annotation>
             <xsd:documentation><![CDATA[
-                If you specify a 'request-channel' you can optionally set the 
-                'requires-reply' attribute as well. If you set this attribute to 
-                'true', a reply must return a non-null value. 
-                
+                If you specify a 'request-channel' you can optionally set the
+                'requires-reply' attribute as well. If you set this attribute to
+                'true', a reply must return a non-null value.
+
                 For example, you dispatch a message to the request-channel
                 (backed by a 'QueueChannel'). If the reply does not return within
-                the specified 'replyTimeout', then the reply message will end up 
+                the specified 'replyTimeout', then the reply message will end up
                 being Null.
-                
+
                 By setting 'requires-reply' to 'true', a 'ReplyRequiredException'
-                will be raised for null reply messages. If 'requires-reply' is set 
-                to false, those messages are silently dropped. 
-                
+                will be raised for null reply messages. If 'requires-reply' is set
+                to false, those messages are silently dropped.
+
                 This attribute defaults to 'true', if not specified.]]>
                 </xsd:documentation>
             </xsd:annotation>
             <xsd:simpleType>
                 <xsd:union memberTypes="xsd:boolean xsd:string" />
-            </xsd:simpleType>            
+            </xsd:simpleType>
         </xsd:attribute>
 		<xsd:attribute name="should-clone-payload">
 			<xsd:annotation>
@@ -1268,17 +1268,28 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="delayer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:sequence minOccurs="0" maxOccurs="1">
+						<xsd:element name="poller" type="basePollerType" />
+					</xsd:sequence>
+					<xsd:attributeGroup ref="inputOutputChannelGroup"/>
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
 	</xsd:element>
 
 	<xsd:complexType name="delayer-type">
-		<xsd:sequence>
-			<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
-		</xsd:sequence>
-		<xsd:attribute name="default-delay" type="xsd:string" use="required">
+		<xsd:attribute name="id" type="xsd:string" use="required">
+			<xsd:annotation>
+				<xsd:documentation>
+					'id' value is used:
+					- as the 'AbstractEndpoint' bean 'id' if this tag is defined as root element.
+					- as the DelayHandler bean alias together with suffix '.handler'
+					- as the 'messageGroupId' property of DelayHandler together with suffix '.messageGroupId'
+					  in the operations of the MessageGroupStore for scheduling delayed messages.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attribute name="default-delay" type="xsd:string">
 			<xsd:annotation>
 				<xsd:documentation>
 					Specify the default delay in milliseconds. This value can be set to 0
@@ -1309,10 +1320,9 @@
 					Provide a reference to the TaskScheduler instance to which
 					this endpoint should
 					delegate when scheduling the sending of delayed Messages. If not
-					provided, the default
-					will use a thread pool of
-					size
-					1.
+					provided, the default scheduler
+					registered in the ApplicationContext {ThreadPoolTaskScheduler} will be
+					used.
 					</xsd:documentation>
 				<xsd:appinfo>
 					<tool:annotation kind="ref">
@@ -1335,14 +1345,6 @@
 				</xsd:appinfo>
 			</xsd:annotation>
 		</xsd:attribute>
-		<xsd:attribute name="wait-for-tasks-to-complete-on-shutdown" type="xsd:string">
-			<xsd:annotation>
-				<xsd:documentation>
-					Specify whether tasks should be able to complete on shutdown. By
-					default this is 'false'.
-				</xsd:documentation>
-			</xsd:annotation>
-		</xsd:attribute>
 	</xsd:complexType>
 
 	<xsd:element name="bridge">
@@ -1358,7 +1360,7 @@
 				<xsd:any processContents="strict" namespace="##other" minOccurs="0" maxOccurs="unbounded" />
 				<xsd:element ref="poller" />
 			</xsd:choice>
-			<xsd:attributeGroup ref="inputOutputChannelGroup" />
+			<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 		</xsd:complexType>
 	</xsd:element>
 
@@ -1372,7 +1374,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="chain-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1635,7 +1637,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="header-enricher-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1895,7 +1897,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="header-filter-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1934,7 +1936,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="expressionOrInnerEndpointDefinitionAware">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1951,7 +1953,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="specialized-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1965,7 +1967,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="specialized-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -1979,7 +1981,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="map-to-object-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2021,7 +2023,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="object-to-json-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2065,7 +2067,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="json-to-object-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2107,7 +2109,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="payload-serializing-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2142,7 +2144,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="payload-deserializing-transformer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2185,7 +2187,7 @@
                 <xsd:sequence minOccurs="0" maxOccurs="1">
                     <xsd:element ref="poller" />
                 </xsd:sequence>
-                <xsd:attributeGroup ref="inputOutputChannelGroup" />
+                <xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
             </xsd:extension>
         </xsd:complexContent>
     </xsd:complexType>
@@ -2213,7 +2215,7 @@
                 <xsd:sequence minOccurs="0" maxOccurs="1">
                     <xsd:element ref="poller" />
                 </xsd:sequence>
-                <xsd:attributeGroup ref="inputOutputChannelGroup" />
+                <xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 	        </xsd:extension>
         </xsd:complexContent>
     </xsd:complexType>
@@ -2279,7 +2281,7 @@
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="filter-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -2987,7 +2989,7 @@ is provided, the return value is expected to match a channel name exactly.
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="splitter-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -3037,7 +3039,7 @@ is provided, the return value is expected to match a channel name exactly.
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="aggregator-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -3188,7 +3190,7 @@ is provided, the return value is expected to match a channel name exactly.
 		<xsd:complexType>
 			<xsd:complexContent>
 				<xsd:extension base="resequencer-type">
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -3604,7 +3606,7 @@ The list of component name patterns you want to track (e.g.,  tracked-components
 					<xsd:all minOccurs="0" maxOccurs="1">
 						<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
 					</xsd:all>
-					<xsd:attributeGroup ref="inputOutputChannelGroup" />
+					<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>
@@ -3625,7 +3627,6 @@ The list of component name patterns you want to track (e.g.,  tracked-components
 	</xsd:complexType>
 
 	<xsd:attributeGroup name="inputOutputChannelGroup">
-		<xsd:attribute name="id" type="xsd:string" />
 		<xsd:attribute name="output-channel" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>
@@ -3681,4 +3682,18 @@ endpoint itself is a Polling Consumer for a channel with a queue.
             </xsd:simpleType>
         </xsd:attribute>
 	</xsd:attributeGroup>
+
+	<xsd:attributeGroup name="inputOutputChannelGroupWithId">
+		<xsd:attribute name="id" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					'id' value:
+					- Identifies the underlying Spring bean definition (AbstractEndpoint)
+					- as MessageHandler bean alias together with suffix '.handler'
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+		<xsd:attributeGroup ref="inputOutputChannelGroup"/>
+	</xsd:attributeGroup>
+
 </xsd:schema>