INT-4471: PubSubChannel: Add errorHandler warn (#2459)

* INT-4471: PubSubChannel: Add errorHandler warn

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

* When an `Executor` is not provided, log warn that the provided
`ErrorHandler` is ignored.

**Cherry-pick to 5.0.x and 4.3.x**

* * Polish warn message

Author: artembilan

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -36,19 +36,18 @@
  */
 public class PublishSubscribeChannel extends AbstractExecutorChannel {
 
-	private volatile ErrorHandler errorHandler;
+	private ErrorHandler errorHandler;
 
-	private volatile boolean ignoreFailures;
+	private boolean ignoreFailures;
 
-	private volatile boolean applySequence;
+	private boolean applySequence;
 
-	private volatile int minSubscribers;
+	private int minSubscribers;
 
 	/**
 	 * Create a PublishSubscribeChannel that will use an {@link Executor}
 	 * to invoke the handlers. If this is null, each invocation will occur in
 	 * the message sender's thread.
-	 *
 	 * @param executor The executor.
 	 */
 	public PublishSubscribeChannel(Executor executor) {
@@ -80,9 +79,7 @@ public String getComponentType() {
 	 * a {@link MessagePublishingErrorHandler} that sends error messages to
 	 * the failed request Message's error channel header if available or to
 	 * the default 'errorChannel' otherwise.
-	 *
 	 * @param errorHandler The error handler.
-	 *
 	 * @see #PublishSubscribeChannel(Executor)
 	 */
 	public void setErrorHandler(ErrorHandler errorHandler) {
@@ -147,6 +144,14 @@ public final void onInit() throws Exception {
 			getDispatcher().setApplySequence(this.applySequence);
 			getDispatcher().setMinSubscribers(this.minSubscribers);
 		}
+		else if (this.errorHandler != null) {
+			if (this.logger.isWarnEnabled()) {
+				this.logger.warn("The 'errorHandler' is ignored for the '" + getComponentName() +
+						"' (an 'executor' is not provided) and exceptions will be thrown " +
+						"directly within the sending Thread");
+			}
+		}
+
 		if (this.maxSubscribers == null) {
 			Integer maxSubscribers =
 					getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);

spring-integration-core/src/test/java/org/springframework/integration/config/publishSubscribeChannelParserTests.xml renamed to spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests-context.xml

@@ -19,7 +19,7 @@
 
 	<publish-subscribe-channel id="channelWithApplySequenceEnabledAndTaskExecutor" apply-sequence="true" task-executor="pool"/>
 
-	<publish-subscribe-channel id="channelWithErrorHandler" error-handler="testErrorHandler"/>
+	<publish-subscribe-channel id="channelWithErrorHandler" error-handler="testErrorHandler" task-executor="pool"/>
 
 	<task:executor id="pool" pool-size="1"/>
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -26,9 +26,11 @@
 import java.util.concurrent.Executor;
 
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.integration.channel.PublishSubscribeChannel;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
 import org.springframework.integration.support.utils.IntegrationUtils;
@@ -37,21 +39,23 @@
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.util.ErrorHandler;
 
 /**
  * @author Mark Fisher
  * @author Gary Russell
  * @author Artem Bilan
  */
+@RunWith(SpringRunner.class)
 public class PublishSubscribeChannelParserTests {
 
+	@Autowired
+	private ApplicationContext context;
+
 	@Test
 	public void defaultChannel() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("defaultChannel");
+		PublishSubscribeChannel channel = this.context.getBean("defaultChannel", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
@@ -69,43 +73,34 @@ public void handleMessage(Message<?> message) throws MessagingException {
 		assertNull(dispatcherAccessor.getPropertyValue("executor"));
 		assertFalse((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
 		assertTrue((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
-		Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
+		Object mbf = this.context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
 		assertSame(mbf, dispatcherAccessor.getPropertyValue("messageBuilderFactory"));
-		context.close();
 	}
 
 	@Test
 	public void ignoreFailures() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithIgnoreFailures");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithIgnoreFailures", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
 		assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("ignoreFailures"));
-		context.close();
 	}
 
 	@Test
 	public void applySequenceEnabled() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithApplySequenceEnabled");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithApplySequenceEnabled", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
 		assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("applySequence"));
-		context.close();
 	}
 
 	@Test
 	public void channelWithTaskExecutor() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithTaskExecutor");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithTaskExecutor", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
@@ -116,15 +111,12 @@ public void channelWithTaskExecutor() {
 		DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
 		Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
 		assertEquals(context.getBean("pool"), innerExecutor);
-		context.close();
 	}
 
 	@Test
 	public void ignoreFailuresWithTaskExecutor() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithIgnoreFailuresAndTaskExecutor");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithIgnoreFailuresAndTaskExecutor", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
@@ -135,16 +127,13 @@ public void ignoreFailuresWithTaskExecutor() {
 		assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
 		DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
 		Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
-		assertEquals(context.getBean("pool"), innerExecutor);
-		context.close();
+		assertEquals(this.context.getBean("pool"), innerExecutor);
 	}
 
 	@Test
 	public void applySequenceEnabledWithTaskExecutor() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithApplySequenceEnabledAndTaskExecutor");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithApplySequenceEnabledAndTaskExecutor", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
 				accessor.getPropertyValue("dispatcher");
@@ -155,21 +144,17 @@ public void applySequenceEnabledWithTaskExecutor() {
 		assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
 		DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
 		Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
-		assertEquals(context.getBean("pool"), innerExecutor);
-		context.close();
+		assertEquals(this.context.getBean("pool"), innerExecutor);
 	}
 
 	@Test
 	public void channelWithErrorHandler() {
-		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"publishSubscribeChannelParserTests.xml", this.getClass());
-		PublishSubscribeChannel channel = (PublishSubscribeChannel)
-				context.getBean("channelWithErrorHandler");
+		PublishSubscribeChannel channel =
+				this.context.getBean("channelWithErrorHandler", PublishSubscribeChannel.class);
 		DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
 		ErrorHandler errorHandler = (ErrorHandler) accessor.getPropertyValue("errorHandler");
 		assertNotNull(errorHandler);
-		assertEquals(context.getBean("testErrorHandler"), errorHandler);
-		context.close();
+		assertEquals(this.context.getBean("testErrorHandler"), errorHandler);
 	}
 
 }

src/reference/asciidoc/channel.adoc

@@ -565,6 +565,10 @@ When using this element, you can also specify the `task-executor` used for publi
 <int:publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>
 ----
 
+Alongside with the `Executor`, an `ErrorHandler` can be configured as well.
+By default the `PublishSubscribeChannel` uses a `MessagePublishingErrorHandler` implementation to send error to the `MessageChannel` from the `errorChannel` header or a global `errorChannel` instance.
+If an `Executor` is not configured, the `ErrorHandler` is ignored and exceptions are thrown directly to the caller's Thread.
+
 If you are providing a _Resequencer_ or _Aggregator_ downstream from a `PublishSubscribeChannel`, then you can set the 'apply-sequence' property on the channel to `true`.
 That will indicate that the channel should set the sequence-size and sequence-number Message headers as well as the correlation id prior to passing the Messages along.
 For example, if there are 5 subscribers, the sequence-size would be set to 5, and the Messages would have sequence-number header values ranging from 1 to 5.