INT-3489 Reschedule Aggregator's group-timeout

JIRA: https://jira.spring.io/browse/INT-3489

* Catch `MessageDeliveryException` and reschedule `group-time out task`: `AbstractCorrelatingMessageHandler#scheduleGroupToForceComplete`
* Apply `send-timeout` for the `discardChannel`
* Mark `groupRemove = false` in case of `MessageDeliveryException`
* Improve `send-timeout` docs

INT-3489: Recalculate `groupTimeout` on each rescheduling

Add Test for Zero Timeout

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -50,6 +50,7 @@ import org.springframework.integration.support.locks.LockRegistry;
 import org.springframework.integration.util.UUIDConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.scheduling.TaskScheduler;
@@ -122,14 +123,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	private volatile boolean expireGroupsUponTimeout = true;
 
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
-									 CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
+			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
 		Assert.notNull(processor);
 
 		Assert.notNull(store);
 		setMessageStore(store);
 		this.outputProcessor = processor;
-		this.correlationStrategy = correlationStrategy == null ?
-				new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID) : correlationStrategy;
+		this.correlationStrategy = (correlationStrategy == null
+				? new HeaderAttributeCorrelationStrategy(IntegrationMessageHeaderAccessor.CORRELATION_ID)
+				: correlationStrategy);
 		this.releaseStrategy = releaseStrategy == null ? new SequenceSizeReleaseStrategy() : releaseStrategy;
 		setSendTimeout(DEFAULT_SEND_TIMEOUT);
 		sequenceAware = this.releaseStrategy instanceof SequenceSizeReleaseStrategy;
@@ -214,9 +216,9 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 		if (this.releasePartialSequences) {
 			Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy,
-					"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName()
-							+ "] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead.");
-			((SequenceSizeReleaseStrategy)this.releaseStrategy).setReleasePartialSequences(releasePartialSequences);
+					"Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() +
+							"] cannot release partial sequences. Use the default SequenceSizeReleaseStrategy instead.");
+			((SequenceSizeReleaseStrategy) this.releaseStrategy).setReleasePartialSequences(releasePartialSequences);
 		}
 
 		/*
@@ -339,7 +341,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 	@Override
 	protected void handleMessageInternal(Message<?> message) throws Exception {
 		Object correlationKey = correlationStrategy.getCorrelationKey(message);
-		Assert.state(correlationKey!=null, "Null correlation not allowed.  Maybe the CorrelationStrategy is failing?");
+		Assert.state(correlationKey != null, "Null correlation not allowed.  Maybe the CorrelationStrategy is failing?");
 
 		if (logger.isDebugEnabled()) {
 			logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
@@ -354,11 +356,12 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 			if (scheduledFuture != null) {
 				boolean canceled = scheduledFuture.cancel(true);
 				if (canceled && logger.isDebugEnabled()) {
-					logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ " + correlationKey + "].");
+					logger.debug("Cancel 'forceComplete' scheduling for MessageGroup with Correlation Key [ "
+							+ correlationKey + "].");
 				}
 			}
 			MessageGroup messageGroup = messageStore.getMessageGroup(correlationKey);
-			if (this.sequenceAware){
+			if (this.sequenceAware) {
 				messageGroup = new SequenceAwareMessageGroup(messageGroup);
 			}
 
@@ -380,29 +383,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 					}
 				}
 				else {
-					Long groupTimeout = this.obtainGroupTimeout(messageGroup);
-					if (groupTimeout != null && groupTimeout >= 0) {
-						if (groupTimeout > 0) {
-							final MessageGroup messageGroupToSchedule = messageGroup;
-
-							scheduledFuture = this.getTaskScheduler()
-									.schedule(new Runnable() {
-
-										@Override
-										public void run() {
-											AbstractCorrelatingMessageHandler.this.forceComplete(messageGroupToSchedule);
-										}
-									}, new Date(System.currentTimeMillis() + groupTimeout));
-
-							if (logger.isDebugEnabled()) {
-								logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'.");
-							}
-							this.expireGroupScheduledFutures.put(groupIdUuid, scheduledFuture);
-						}
-						else {
-							this.forceComplete(messageGroup);
-						}
-					}
+					scheduleGroupToForceComplete(messageGroup);
 				}
 			}
 			else {
@@ -414,6 +395,43 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		}
 	}
 
+	private void scheduleGroupToForceComplete(final MessageGroup messageGroup) {
+		final Long groupTimeout = this.obtainGroupTimeout(messageGroup);
+		/*
+		 * When 'groupTimeout' is evaluated to 'null' we do nothing.
+		 * The 'MessageGroupStoreReaper' can be used to 'forceComplete' message groups.
+		 */
+		if (groupTimeout != null && groupTimeout >= 0) {
+			if (groupTimeout > 0) {
+				ScheduledFuture<?> scheduledFuture = this.getTaskScheduler()
+						.schedule(new Runnable() {
+
+							@Override
+							public void run() {
+								try {
+									AbstractCorrelatingMessageHandler.this.forceComplete(messageGroup);
+								}
+								catch (MessageDeliveryException e) {
+									if (logger.isDebugEnabled()) {
+										logger.debug("The MessageGroup [ " + messageGroup +
+												"] is rescheduled by the reason: " + e.getMessage());
+									}
+									scheduleGroupToForceComplete(messageGroup);
+								}
+							}
+						}, new Date(System.currentTimeMillis() + groupTimeout));
+
+				if (logger.isDebugEnabled()) {
+					logger.debug("Schedule MessageGroup [ " + messageGroup + "] to 'forceComplete'.");
+				}
+				this.expireGroupScheduledFutures.put(UUIDConverter.getUUID(messageGroup.getGroupId()), scheduledFuture);
+			}
+			else {
+				forceComplete(messageGroup);
+			}
+		}
+	}
+
 	private void discardMessage(Message<?> message) {
 		if (this.discardChannelName != null) {
 			synchronized (this) {
@@ -429,7 +447,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 				}
 			}
 		}
-		this.discardChannel.send(message);
+		this.messagingTemplate.send(this.discardChannel, message);
 	}
 
 	/**
@@ -448,7 +466,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		try {
 			lock.lockInterruptibly();
 			try {
-				ScheduledFuture<?> scheduledFuture = this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey));
+				ScheduledFuture<?> scheduledFuture =
+						this.expireGroupScheduledFutures.remove(UUIDConverter.getUUID(correlationKey));
 				if (scheduledFuture != null) {
 					boolean canceled = scheduledFuture.cancel(false);
 					if (canceled && logger.isDebugEnabled()) {
@@ -512,6 +531,15 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 					}
 				}
 			}
+			catch (MessageDeliveryException e) {
+				removeGroup = false;
+				if (logger.isDebugEnabled()) {
+					logger.debug("Group expiry candidate (" + correlationKey +
+							") has been affected by MessageDeliveryException - " +
+							"it may be reconsidered for a future expiration one more time");
+				}
+				throw e;
+			}
 			finally {
 				try {
 					if (removeGroup) {
@@ -534,7 +562,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		messageStore.removeMessageGroup(correlationKey);
 	}
 
-	protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence){
+	protected int findLastReleasedSequenceNumber(Object groupId, Collection<Message<?>> partialSequence) {
 		List<Message<?>> sorted = new ArrayList<Message<?>>(partialSequence);
 		Collections.sort(sorted, new SequenceNumberComparator());
 
@@ -570,7 +598,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		}
 		if (this.applicationEventPublisher != null) {
 			this.applicationEventPublisher.publishEvent(new MessageGroupExpiredEvent(this, correlationKey, group
-					.size(), new Date(group.getLastModified()) , new Date(), !sendPartialResultOnExpiry));
+					.size(), new Date(group.getLastModified()), new Date(), !sendPartialResultOnExpiry));
 		}
 	}
 
@@ -598,9 +626,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		return partialSequence;
 	}
 
-	protected void verifyResultCollectionConsistsOfMessages(Collection<?> elements){
+	protected void verifyResultCollectionConsistsOfMessages(Collection<?> elements) {
 		Class<?> commonElementType = CollectionUtils.findCommonElementType(elements);
-		Assert.isAssignable(Message.class, commonElementType, "The expected collection of Messages contains non-Message element: " + commonElementType);
+		Assert.isAssignable(Message.class, commonElementType,
+				"The expected collection of Messages contains non-Message element: " + commonElementType);
 	}
 
 	@SuppressWarnings("rawtypes")
@@ -619,7 +648,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 			for (Object next : (Iterable<?>) processorResult) {
 				this.sendReplyMessage(next, replyChannel);
 			}
-		} else {
+		}
+		else {
 			this.sendReplyMessage(processorResult, replyChannel);
 		}
 	}
@@ -628,16 +658,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		if (replyChannel instanceof MessageChannel) {
 			if (reply instanceof Message<?>) {
 				this.messagingTemplate.send((MessageChannel) replyChannel, (Message<?>) reply);
-			} else {
+			}
+			else {
 				this.messagingTemplate.convertAndSend((MessageChannel) replyChannel, reply);
 			}
-		} else if (replyChannel instanceof String) {
+		}
+		else if (replyChannel instanceof String) {
 			if (reply instanceof Message<?>) {
 				this.messagingTemplate.send((String) replyChannel, (Message<?>) reply);
-			} else {
+			}
+			else {
 				this.messagingTemplate.convertAndSend((String) replyChannel, reply);
 			}
-		} else {
+		}
+		else {
 			throw new MessagingException("replyChannel must be a MessageChannel or String");
 		}
 	}

spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouterManagement.java

@@ -53,6 +53,7 @@ public interface RecipientListRouterManagement {
 	/**
 	 * Remove all recipients that match the channelName.
 	 * @param channelName The channel name.
+	 * @return The number of recipients removed.
 	 */
 	@ManagedOperation
 	int removeRecipient(String channelName);
@@ -61,6 +62,7 @@ public interface RecipientListRouterManagement {
 	 * Remove all recipients that match the channelName and expression.
 	 * @param channelName The channel name.
 	 * @param selectorExpression The expression to filter the incoming message
+	 * @return The number of recipients removed.
 	 */
 	@ManagedOperation
 	int removeRecipient(String channelName, String selectorExpression);

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

@@ -4295,8 +4295,10 @@ The list of component name patterns you want to track (e.g.,  tracked-components
 			<xsd:annotation>
 				<xsd:documentation>
 					Specify the maximum amount of time in milliseconds to wait when sending a reply
-					Message to the
-					output channel. By default the send will block for one second.
+					Message to the output channel. By default the send will block for one second.
+					It is applied only if the output channel has some 'sending' limitations, e.g. QueueChannel with
+					fixed a 'capacity'. In this case a MessageDeliveryException is thrown. The 'send-timeout'
+					is ignored in case of AbstractSubscribableChannel implementations.
 				</xsd:documentation>
 			</xsd:annotation>
 		</xsd:attribute>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerIntegrationTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.aggregator;
 
+import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.isA;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -63,6 +64,7 @@ public class CorrelatingMessageHandlerIntegrationTests {
 	public void completesAfterThreshold() throws Exception {
 		defaultHandler.setReleaseStrategy(new MessageCountReleaseStrategy());
 		MessageChannel discardChannel = mock(MessageChannel.class);
+		when(discardChannel.send(any(Message.class))).thenReturn(true);
 		defaultHandler.setDiscardChannel(discardChannel);
 		Message<?> message1 = correlatedMessage(1, 2, 1);
 		Message<?> message2 = correlatedMessage(1, 2, 2);

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests-context.xml

@@ -37,6 +37,7 @@
 	<aggregator id="gta"
 				input-channel="groupTimeoutAggregatorInput" output-channel="output" discard-channel="discard"
 				send-partial-result-on-expiry="true"
+				send-timeout="100"
 				group-timeout="100"/>
 
 	<aggregator input-channel="groupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
@@ -44,5 +45,18 @@
 				group-timeout-expression="size() ge 2 ? 100 : -1"
 			    release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
 
+	<aggregator input-channel="zeroGroupTimeoutExpressionAggregatorInput" output-channel="output" discard-channel="discard"
+				send-partial-result-on-expiry="true"
+				send-timeout="10"
+				group-timeout-expression="@bool.getAndSet(true) ? 0 : 10"
+			    release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/>
+
+	<beans:bean id="bool" class="java.util.concurrent.atomic.AtomicBoolean" />
+
+	<bridge input-channel="errorChannel" output-channel="errors" />
+
+	<channel id="errors">
+		<queue />
+	</channel>
 
 </beans:beans>

spring-integration-core/src/test/java/org/springframework/integration/aggregator/integration/AggregatorIntegrationTests.java

@@ -16,15 +16,19 @@
 
 package org.springframework.integration.aggregator.integration;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -32,11 +36,13 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.store.MessageGroupStore;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -71,11 +77,17 @@ public class AggregatorIntegrationTests {
 	private MessageChannel groupTimeoutExpressionAggregatorInput;
 
 	@Autowired
-	private PollableChannel output;
+	private MessageChannel zeroGroupTimeoutExpressionAggregatorInput;
+
+	@Autowired
+	private QueueChannel output;
 
 	@Autowired
 	private PollableChannel discard;
 
+	@Autowired
+	private QueueChannel errors;
+
 	@Test//(timeout=5000)
 	public void testVanillaAggregation() throws Exception {
 		for (int i = 0; i < 5; i++) {
@@ -147,6 +159,31 @@ public class AggregatorIntegrationTests {
 		}
 	}
 
+	@Test
+	public void testGroupTimeoutReschedulingOnMessageDeliveryException() throws Exception {
+		for (int i = 0; i < 5; i++) {
+			this.output.send(new GenericMessage<String>("fake message"));
+		}
+
+		Map<String, Object> headers = stubHeaders(1, 2, 1);
+		this.groupTimeoutAggregatorInput.send(new GenericMessage<Integer>(1, headers));
+
+		//Wait until 'group-timeout' does its stuff.
+		MessageGroupStore mgs = TestUtils.getPropertyValue(this.context.getBean("gta.handler"), "messageStore",
+				MessageGroupStore.class);
+		int n = 0;
+		while (n++ < 100 && mgs.getMessageGroupCount() > 0) {
+			Thread.sleep(100);
+			if (n == 10) {
+				TestUtils.getPropertyValue(this.output, "queue", Queue.class).clear();
+			}
+		}
+		assertTrue("Group did not complete", n < 100);
+		Message<?> receive = this.output.receive(1000);
+		assertNotNull(receive);
+		assertEquals(Collections.singletonList(1), receive.getPayload());
+		assertNull(this.discard.receive(0));
+	}
 
 	@Test
 	public void testGroupTimeoutExpressionScheduling() throws Exception {
@@ -187,6 +224,28 @@ public class AggregatorIntegrationTests {
 		assertNull(this.discard.receive(0));
 	}
 
+	@Test
+	public void testZeroGroupTimeoutExpressionScheduling() throws Exception {
+		try {
+			this.output.purge(null);
+			this.errors.purge(null);
+			GenericMessage<String> message = new GenericMessage<String>("foo");
+			this.output.send(message);
+			this.output.send(message);
+			this.output.send(message);
+			this.output.send(message);
+			this.output.send(message);
+			this.zeroGroupTimeoutExpressionAggregatorInput.send(new GenericMessage<Integer>(1, stubHeaders(1, 2, 1)));
+			ErrorMessage em = (ErrorMessage) this.errors.receive(10000);
+			assertNotNull(em);
+			assertThat(em.getPayload().getMessage(),
+					containsString("failed to send message to channel 'output' within timeout: 10"));
+		}
+		finally {
+			this.output.purge(null);
+			this.errors.purge(null);
+		}
+	}
 
 	// configured in context associated with this test
 	public static class SummingAggregator {

spring-integration-core/src/test/java/org/springframework/integration/json/ObjectToJsonTransformerTests.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.
@@ -53,21 +53,21 @@ public class ObjectToJsonTransformerTests {
 
 	@Test
 	public void simpleStringPayload() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		String result = (String) transformer.transform(new GenericMessage<String>("foo")).getPayload();
 		assertEquals("\"foo\"", result);
 	}
 
 	@Test
 	public void withDefaultContentType() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		Message<?> result = transformer.transform(new GenericMessage<String>("foo"));
 		assertEquals(ObjectToJsonTransformer.JSON_CONTENT_TYPE, result.getHeaders().get(MessageHeaders.CONTENT_TYPE));
 	}
 
 	@Test
 	public void withProvidedContentType() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build();
 		Message<?> result = transformer.transform(message);
 		assertEquals("text/xml", result.getHeaders().get(MessageHeaders.CONTENT_TYPE));
@@ -75,7 +75,7 @@ public class ObjectToJsonTransformerTests {
 
 	@Test
 	public void withProvidedContentTypeWithOverride() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		transformer.setContentType(ObjectToJsonTransformer.JSON_CONTENT_TYPE);
 		Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build();
 		Message<?> result = transformer.transform(message);
@@ -84,7 +84,7 @@ public class ObjectToJsonTransformerTests {
 
 	@Test
 	public void withProvidedContentTypeAsEmptyString() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		transformer.setContentType("");
 		Message<?> message = MessageBuilder.withPayload("foo").build();
 		Message<?> result = transformer.transform(message);
@@ -93,29 +93,29 @@ public class ObjectToJsonTransformerTests {
 
 	@Test
 	public void withProvidedContentTypeAsEmptyStringDoesNotOverride() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		transformer.setContentType("");
 		Message<?> message = MessageBuilder.withPayload("foo").setHeader(MessageHeaders.CONTENT_TYPE, "text/xml").build();
 		Message<?> result = transformer.transform(message);
 		assertEquals("text/xml", result.getHeaders().get(MessageHeaders.CONTENT_TYPE));
 	}
 
-	@Test(expected=IllegalArgumentException.class)
+	@Test(expected = IllegalArgumentException.class)
 	public void withProvidedContentTypeAsNull() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		transformer.setContentType(null);
 	}
 
 	@Test
 	public void simpleIntegerPayload() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		String result = (String) transformer.transform(new GenericMessage<Integer>(123)).getPayload();
 		assertEquals("123", result);
 	}
 
 	@Test
 	public void objectPayload() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer();
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer();
 		TestAddress address = new TestAddress(123, "Main Street");
 		TestPerson person = new TestPerson("John", "Doe", 42);
 		person.setAddress(address);
@@ -135,7 +135,7 @@ public class ObjectToJsonTransformerTests {
 	public void objectPayloadWithCustomObjectMapper() throws Exception {
 		ObjectMapper customMapper = new ObjectMapper();
 		customMapper.configure(JsonGenerator.Feature.QUOTE_FIELD_NAMES, Boolean.FALSE);
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer(new Jackson2JsonObjectMapper(customMapper));
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new Jackson2JsonObjectMapper(customMapper));
 		TestPerson person = new TestPerson("John", "Doe", 42);
 		person.setAddress(new TestAddress(123, "Main Street"));
 		String result = (String) transformer.transform(new GenericMessage<TestPerson>(person)).getPayload();
@@ -152,7 +152,7 @@ public class ObjectToJsonTransformerTests {
 
 	@Test
 	public void testBoonJsonObjectMapper() throws Exception {
-		ObjectToJsonTransformer transformer = new  ObjectToJsonTransformer(new BoonJsonObjectMapper());
+		ObjectToJsonTransformer transformer = new ObjectToJsonTransformer(new BoonJsonObjectMapper());
 		TestPerson person = new TestPerson("John", "Doe", 42);
 		person.setAddress(new TestAddress(123, "Main Street"));
 		String result = (String) transformer.transform(new GenericMessage<TestPerson>(person)).getPayload();

src/reference/docbook/aggregator.xml

@@ -452,8 +452,15 @@ then you should simply provide an implementation of the <classname>ReleaseStrate
       </callout>
 
        <callout arearefs="aggxml09" id="aggxml09-txt">
-        <para>The timeout interval for sending the aggregated messages to the output or
-        reply channel. <emphasis>Optional</emphasis>.</para>
+        <para>The timeout interval to wait when sending a reply
+			<interfacename>Message</interfacename> to the <code>output-channel</code> or <code>discard-channel</code>.
+			By default the send will block for one second.
+			It is applied only if the output channel has some 'sending' limitations, e.g. <classname>QueueChannel</classname>
+			with a fixed 'capacity'. In this case a <classname>MessageDeliveryException</classname> is thrown.
+			The <code>send-timeout</code> is ignored in case of <classname>AbstractSubscribableChannel</classname> implementations.
+			In case of <code>group-timeout(-expression)</code> the <classname>MessageDeliveryException</classname>
+			from the scheduled expire task leads this task to be rescheduled.
+			<emphasis>Optional</emphasis>.</para>
       </callout>
 
        <callout arearefs="aggxml10" id="aggxml10-txt">

src/reference/docbook/resequencer.xml

@@ -102,8 +102,15 @@
         </callout>
 
         <callout arearefs="resxml10-co" id="resxml10">
-          <para>The timeout for sending out messages.
-          <emphasis>Optional</emphasis>.</para>
+		<para>The timeout interval to wait when sending a reply
+			<interfacename>Message</interfacename> to the <code>output-channel</code> or <code>discard-channel</code>.
+			By default the send will block for one second.
+			It is applied only if the output channel has some 'sending' limitations, e.g. <classname>QueueChannel</classname>
+			with a fixed 'capacity'. In this case a <classname>MessageDeliveryException</classname> is thrown.
+			The <code>send-timeout</code> is ignored in case of <classname>AbstractSubscribableChannel</classname> implementations.
+			In case of <code>group-timeout(-expression)</code> the <classname>MessageDeliveryException</classname>
+			from the scheduled expire task leads this task to be rescheduled.
+			<emphasis>Optional</emphasis>.</para>
         </callout>
 
         <callout arearefs="resxml11-co" id="resxml11">