INT-4113: Add @Poller.errorChannel() Attribute

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

* To simplify `MessagePublishingErrorHandler` and avoid extra `PollerMetadata` beans, added the `errorChannel()` attribute to the `@Poller` annotation
* The `MessagePublishingErrorHandler` now supports the late-binding via new `defaultErrorChannelName`
* Docs about new `errorChannel()` attribute
* Some other docs polishing

Polishing

**Cherry-pick to 4.3.x**

Polishing docs according PR comments

spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java

@@ -85,4 +85,11 @@ public @interface Poller {
 	 */
 	String cron() default "";
 
+	/**
+	 * @return The the bean name of default error channel
+	 * for the underlying {@code MessagePublishingErrorHandler}.
+	 * @since 4.3.3
+	 */
+	String errorChannel() default "";
+
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/MessagePublishingErrorHandler.java

@@ -39,6 +39,7 @@ import org.springframework.util.ErrorHandler;
  * @author Iwein Fuld
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryAware {
 
@@ -48,6 +49,8 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 
 	private volatile MessageChannel defaultErrorChannel;
 
+	private volatile String defaultErrorChannelName;
+
 	private volatile long sendTimeout = 1000;
 
 
@@ -70,9 +73,25 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 	 * @since 4.3
 	 */
 	public MessageChannel getDefaultErrorChannel() {
+		String defaultErrorChannelName = this.defaultErrorChannelName;
+		if (defaultErrorChannelName != null) {
+			if (this.channelResolver != null) {
+				this.defaultErrorChannel = this.channelResolver.resolveDestination(defaultErrorChannelName);
+				this.defaultErrorChannelName = null;
+			}
+		}
 		return this.defaultErrorChannel;
 	}
 
+	/**
+	 * Specify the bean name of default error channel for this error handler.
+	 * @param defaultErrorChannelName the bean name of the error channel
+	 * @since 4.3.3
+	 */
+	public void setDefaultErrorChannelName(String defaultErrorChannelName) {
+		this.defaultErrorChannelName = defaultErrorChannelName;
+	}
+
 	public void setSendTimeout(long sendTimeout) {
 		this.sendTimeout = sendTimeout;
 	}
@@ -87,7 +106,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 
 	@Override
 	public final void handleError(Throwable t) {
-		MessageChannel errorChannel = this.resolveErrorChannel(t);
+		MessageChannel errorChannel = resolveErrorChannel(t);
 		boolean sent = false;
 		if (errorChannel != null) {
 			try {
@@ -123,7 +142,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 	private MessageChannel resolveErrorChannel(Throwable t) {
 		Message<?> failedMessage = (t instanceof MessagingException) ?
 				((MessagingException) t).getFailedMessage() : null;
-		if (this.defaultErrorChannel == null && this.channelResolver != null) {
+		if (getDefaultErrorChannel() == null && this.channelResolver != null) {
 			this.defaultErrorChannel = this.channelResolver.resolveDestination(
 					IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME);
 		}
@@ -137,7 +156,7 @@ public class MessagePublishingErrorHandler implements ErrorHandler, BeanFactoryA
 		}
 		Assert.isInstanceOf(String.class, errorChannelHeader,
 				"Unsupported error channel header type. Expected MessageChannel or String, but actual type is [" +
-				errorChannelHeader.getClass() + "]");
+						errorChannelHeader.getClass() + "]");
 		return this.channelResolver.resolveDestination((String) errorChannelHeader);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -49,6 +49,7 @@ import org.springframework.core.task.TaskExecutor;
 import org.springframework.integration.annotation.IdempotentReceiver;
 import org.springframework.integration.annotation.Poller;
 import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.config.IntegrationConfigUtils;
 import org.springframework.integration.context.Orderable;
 import org.springframework.integration.endpoint.AbstractEndpoint;
@@ -236,8 +237,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		boolean createEndpoint = StringUtils.hasText(inputChannel);
 		if (!createEndpoint && beanAnnotationAware()) {
 			boolean isBean = AnnotatedElementUtils.isAnnotated(method, Bean.class.getName());
-			Assert.isTrue(!isBean, "A channel name in '" + getInputChannelAttribute() + "' is required when " + this.annotationType +
-					" is used on '@Bean' methods.");
+			Assert.isTrue(!isBean, "A channel name in '" + getInputChannelAttribute() + "' is required when " +
+					this.annotationType + " is used on '@Bean' methods.");
 		}
 		return createEndpoint;
 	}
@@ -343,7 +344,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		PollerMetadata pollerMetadata = null;
 		Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
 		if (!ObjectUtils.isEmpty(pollers)) {
-			Assert.state(pollers.length == 1, "The 'poller' for an Annotation-based endpoint can have only one '@Poller'.");
+			Assert.state(pollers.length == 1,
+					"The 'poller' for an Annotation-based endpoint can have only one '@Poller'.");
 			Poller poller = pollers[0];
 
 			String ref = poller.value();
@@ -353,6 +355,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			String fixedRateValue = this.beanFactory.resolveEmbeddedValue(poller.fixedRate());
 			String maxMessagesPerPollValue = this.beanFactory.resolveEmbeddedValue(poller.maxMessagesPerPoll());
 			String cron = this.beanFactory.resolveEmbeddedValue(poller.cron());
+			String errorChannel = this.beanFactory.resolveEmbeddedValue(poller.errorChannel());
 
 			if (StringUtils.hasText(ref)) {
 				Assert.state(!StringUtils.hasText(triggerRef) && !StringUtils.hasText(executorRef) &&
@@ -370,9 +373,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 					// SPCAs default to 1 message per poll
 					pollerMetadata.setMaxMessagesPerPoll(1);
 				}
+
 				if (StringUtils.hasText(executorRef)) {
 					pollerMetadata.setTaskExecutor(this.beanFactory.getBean(executorRef, TaskExecutor.class));
 				}
+
 				Trigger trigger = null;
 				if (StringUtils.hasText(triggerRef)) {
 					Assert.state(!StringUtils.hasText(cron) && !StringUtils.hasText(fixedDelayValue)
@@ -396,6 +401,13 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 				}
 				//'Trigger' can be null. 'PollingConsumer' does fallback to the 'new PeriodicTrigger(10)'.
 				pollerMetadata.setTrigger(trigger);
+
+				if (StringUtils.hasText(errorChannel)) {
+					MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
+					errorHandler.setDefaultErrorChannelName(errorChannel);
+					errorHandler.setBeanFactory(this.beanFactory);
+					pollerMetadata.setErrorHandler(errorHandler);
+				}
 			}
 		}
 		else {
@@ -425,8 +437,10 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 		return name + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX;
 	}
 
-	protected void setOutputChannelIfPresent(List<Annotation> annotations, AbstractReplyProducingMessageHandler handler) {
-		String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel", String.class);
+	protected void setOutputChannelIfPresent(List<Annotation> annotations,
+			AbstractReplyProducingMessageHandler handler) {
+		String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "outputChannel",
+				String.class);
 		if (StringUtils.hasText(outputChannelName)) {
 			handler.setOutputChannelName(outputChannelName);
 		}

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java

@@ -45,6 +45,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;
 import org.springframework.expression.ExpressionParser;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.MessageRejectedException;
 import org.springframework.integration.aggregator.AggregatingMessageHandler;
 import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
 import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy;
@@ -75,6 +76,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
@@ -119,6 +121,9 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 	@Qualifier("skippedMessageSource")
 	private MessageSource<?> skippedMessageSource;
 
+	@Autowired
+	private PollableChannel counterErrorChannel;
+
 	@Test
 	public void testMessagingAnnotationsFlow() {
 		this.sourcePollingChannelAdapter.start();
@@ -126,6 +131,17 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 			Message<?> receive = this.discardChannel.receive(10000);
 			assertNotNull(receive);
 			assertTrue(((Integer) receive.getPayload()) % 2 == 0);
+
+			receive = this.counterErrorChannel.receive(10000);
+			assertNotNull(receive);
+			assertThat(receive, instanceOf(ErrorMessage.class));
+			assertThat(receive.getPayload(), instanceOf(MessageRejectedException.class));
+			MessageRejectedException exception = (MessageRejectedException) receive.getPayload();
+			assertThat(exception.getMessage(),
+					containsString("MessageFilter " +
+							"'messagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.filter.filter.handler'" +
+							" rejected Message"));
+
 		}
 		for (Message<?> message : collector) {
 			assertFalse(((Integer) message.getPayload()) % 2 == 0);
@@ -174,15 +190,14 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 
 		@Bean
 		@InboundChannelAdapter(value = "routerChannel", autoStartup = "false",
-				poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
+				poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1", errorChannel = "counterErrorChannel"))
 		public MessageSource<Integer> counterMessageSource(final AtomicInteger counter) {
-			return new MessageSource<Integer>() {
+			return () -> new GenericMessage<>(counter.incrementAndGet());
+		}
 
-				@Override
-				public Message<Integer> receive() {
-					return new GenericMessage<Integer>(counter.incrementAndGet());
-				}
-			};
+		@Bean
+		public PollableChannel counterErrorChannel() {
+			return new QueueChannel();
 		}
 
 		@Bean
@@ -191,7 +206,7 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 		}
 
 		@Bean
-		@Router(inputChannel = "routerChannel", channelMappings = {"true=odd", "false=filter"}, suffix = "Channel")
+		@Router(inputChannel = "routerChannel", channelMappings = { "true=odd", "false=filter" }, suffix = "Channel")
 		public MessageSelector router() {
 			return new ExpressionEvaluatingSelector("payload % 2 == 0");
 		}
@@ -208,7 +223,10 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 		}
 
 		@Bean
-		@Filter(inputChannel = "filterChannel", outputChannel = "aggregatorChannel", discardChannel = "discardChannel")
+		@Filter(inputChannel = "filterChannel",
+				outputChannel = "aggregatorChannel",
+				discardChannel = "discardChannel",
+				throwExceptionOnRejection = "true")
 		public MessageSelector filter() {
 			return new ExpressionEvaluatingSelector("payload % 2 != 0");
 		}
@@ -277,7 +295,8 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 		@Filter(inputChannel = "skippedChannel5")
 		@Profile("foo")
 		public MessageHandler skippedMessageHandler() {
-			return m -> { };
+			return m -> {
+			};
 		}
 
 		@Bean

src/reference/asciidoc/configuration.adoc

@@ -374,7 +374,7 @@ public class AnnotationService {
 
 This annotation provides only simple `PollerMetadata` options.
 The `@Poller`'s attributes `maxMessagesPerPoll`, `fixedDelay`, `fixedRate` and `cron` can be configured with _property-placeholders_.
-If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler), the `PollerMetadata` should be configured as a generic bean with its bean name used for `@Poller`'s `value` attribute.
+If it is necessary to provide more polling options (e.g. transaction, advice-chain, error-handler etc.), the `PollerMetadata` should be configured as a generic bean with its bean name used for `@Poller`'s `value` attribute.
 In this case, no other attributes are allowed (they would be specified on the `PollerMetadata` bean).
 Note, if `inputChannel` is `PollableChannel` and no `@Poller` is configured, the default `PollerMetadata` will be used, if it is present in the application context.
 To declare the default poller using `@Configuration`, use:
@@ -424,6 +424,10 @@ public class AnnotationService {
 }
 ----
 
+Starting with _version 4.3.3_, the `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`.
+This attribute play the same role as `error-channel` in the `<poller>` xml component.
+See <<endpoint-namespace>> for more information.
+
 *@InboundChannelAdapter*
 
 Starting with _version 4.0_, the `@InboundChannelAdapter` method annotation is available.

src/reference/asciidoc/endpoint.adoc

@@ -152,14 +152,14 @@ For example, an _Aggregator_ waits for a number of Messages to arrive and is oft
 When using the namespace configuration, you do not strictly need to know all of the details, but it still might be worth knowing that several of these components share a common base class, the `AbstractReplyProducingMessageHandler`, and it provides a `setOutputChannel(..)` method.
 
 [[endpoint-namespace]]
-==== Namespace Support
+==== Endpoint Namespace Support
 
 Throughout the reference manual, you will see specific configuration examples for endpoint elements, such as router, transformer, service-activator, and so on.
 Most of these will support an _input-channel_ attribute and many will support an _output-channel_ attribute.
 After being parsed, these endpoint elements produce an instance of either the `PollingConsumer` or the `EventDrivenConsumer` depending on the type of the _input-channel_ that is referenced: `PollableChannel` or `SubscribableChannel` respectively.
 When the channel is pollable, then the polling behavior is determined based on the endpoint element's _poller_ sub-element and its attributes.
 
-_Configuration_Below you find a _poller_ with all available configuration options:
+In the configuration below you find a _poller_ with all available configuration options:
 
 [source,xml]
 ----

src/reference/asciidoc/mail.adoc

@@ -188,7 +188,7 @@ Alternatively, provide the host, username, and password:
 ----
 
 NOTE: Keep in mind, as with any outbound Channel Adapter, if the referenced channel is a `PollableChannel`,
-a `<poller>` sub-element should be provided (see ).
+a `<poller>` sub-element should be provided (see <<endpoint-namespace>>).
 
 When using the namespace support, a _header-enricher_ Message Transformer is also available.
 This simplifies the application of the headers mentioned above to any Message prior to sending to the Mail Outbound Channel Adapter.

src/reference/asciidoc/whats-new.adoc

@@ -15,6 +15,9 @@ development process.
 
 ==== Core Changes
 
+The `@Poller` annotation now has the `errorChannel` attribute for easier configuration of the underlying `MessagePublishingErrorHandler`.
+See <<annotations>> for more information.
+
 ==== JMS Changes
 
 Previously, Spring Integration JMS XML configuration used a default bean name `connectionFactory` for the JMS Connection Factory, allowing the property to be omitted from component definitions.