Added dispatcher error handling to the message bus and refactored UnicastMessageDispatcher to properly iterate until success or rejection limit is reached.

spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java

@@ -33,9 +33,9 @@ public class ConsumerPolicy {
 
 	private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 1;
 
-	private static final int DEFAULT_REJECTION_LIMIT = 10;
+	private static final int DEFAULT_REJECTION_LIMIT = 5;
 
-	private static final int DEFAULT_REJECTION_LIMIT_WAIT = 1000;
+	private static final int DEFAULT_RETRY_INTERVAL = 1000;
 
 	private static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
@@ -48,7 +48,7 @@ public class ConsumerPolicy {
 
 	private int rejectionLimit = DEFAULT_REJECTION_LIMIT;
 
-	private int rejectionLimitWait = DEFAULT_REJECTION_LIMIT_WAIT;
+	private int retryInterval = DEFAULT_RETRY_INTERVAL;
 
 	private int initialDelay = 0;
 
@@ -82,7 +82,7 @@ public class ConsumerPolicy {
 		ConsumerPolicy policy = new ConsumerPolicy();
 		policy.setPeriod(-1);
 		policy.setConcurrency(1);
-		policy.setMaxConcurrency(1);
+		policy.setMaxConcurrency(10);
 		return policy;
 	}
 
@@ -157,17 +157,17 @@ public class ConsumerPolicy {
 
 	public void setRejectionLimit(int rejectionLimit) {
 		if (rejectionLimit < 1) {
-			throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1");
+			throw new IllegalArgumentException("'rejectionLimit' must be at least 1");
 		}
 		this.rejectionLimit = rejectionLimit;
 	}
 
-	public int getRejectionLimitWait() {
-		return this.rejectionLimitWait;
+	public int getRetryInterval() {
+		return this.retryInterval;
 	}
 
-	public void setRejectionLimitWait(int rejectionLimitWait) {
-		this.rejectionLimitWait = rejectionLimitWait;
+	public void setRetryInterval(int retryInterval) {
+		this.retryInterval = retryInterval;
 	}
 
 	public long getReceiveTimeout() {

spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -38,7 +38,9 @@ import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.MessageReceiver;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -66,6 +68,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 
 	private ScheduledThreadPoolExecutor dispatcherExecutor;
 
+	private int dispatcherPoolSize = 10;
+
 	private boolean autoCreateChannels;
 
 	private boolean running;
@@ -82,6 +86,21 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		this.activateSubscriptions(applicationContext);
 	}
 
+	/**
+	 * Set the size for the dispatcher thread pool.
+	 */
+	public void setDispatcherPoolSize(int dispatcherPoolSize) {
+		Assert.isTrue(dispatcherPoolSize > 0, "'dispatcherPoolSize' must be at least 1");
+		this.dispatcherPoolSize = dispatcherPoolSize;
+		if (this.dispatcherExecutor != null) {
+			this.dispatcherExecutor.setCorePoolSize(dispatcherPoolSize);
+		}
+	}
+
+	/**
+	 * Set whether the bus should automatically create a channel when a
+	 * subscription contains the name of a previously unregistered channel.
+	 */
 	public void setAutoCreateChannels(boolean autoCreateChannels) {
 		this.autoCreateChannels = autoCreateChannels;
 	}
@@ -136,7 +155,17 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	}
 
 	public void initialize() {
-		this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1);
+		initDispatcherExecutor();
+		if (this.getInvalidMessageChannel() == null) {
+			this.setInvalidMessageChannel(new PointToPointChannel(Integer.MAX_VALUE));
+		}
+	}
+
+	private void initDispatcherExecutor() {
+		CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
+		threadFactory.setThreadNamePrefix("dispatcher-executor-");
+		threadFactory.setThreadGroup(new ThreadGroup("dispatcher-executors"));
+		this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherPoolSize, threadFactory);
 	}
 
 	public MessageChannel getInvalidMessageChannel() {
@@ -179,7 +208,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 	}
 
 	public void registerSourceAdapter(String name, SourceAdapter adapter) {
-		// TODO: use the name
 		if (adapter instanceof MessageDispatcher) {
 			MessageDispatcher dispatcher = (MessageDispatcher) adapter;
 			ConsumerPolicy policy = dispatcher.getConsumerPolicy();
@@ -348,8 +376,19 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 	}
 
+	private void handleDispatchError(DispatcherTask task, Throwable t) {
+		try {
+			this.getInvalidMessageChannel().send(new ErrorMessage(t), 1000);
+		}
+		catch (Throwable ignore) { // message will be logged only
+		}
+		if (logger.isWarnEnabled()) {
+			logger.warn("failure occurred while dispatching message", t);
+		}
+	}
 
-	private static class DispatcherTask implements Runnable {
+
+	private class DispatcherTask implements Runnable {
 
 		private MessageDispatcher dispatcher;
 
@@ -367,7 +406,12 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		}
 
 		public void run() {
-			dispatcher.dispatch();
+			try {
+				dispatcher.dispatch();
+			}
+			catch (Throwable t) {
+				handleDispatchError(this, t);
+			}
 		}
 
 	}

spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java

@@ -19,6 +19,7 @@ package org.springframework.integration.bus;
 import java.util.Iterator;
 import java.util.concurrent.RejectedExecutionException;
 
+import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.message.Message;
 
 /**
@@ -45,47 +46,50 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher {
 	@Override
 	protected boolean dispatchMessage(Message<?> message) {
 		int attempts = 0;
-		Iterator<MessageReceivingExecutor> iter = this.getExecutors().iterator();
-		if (!iter.hasNext()) {
-			if (logger.isWarnEnabled()) {
-				logger.warn("dispatcher has no active executors");
+		while (attempts < policy.getRejectionLimit()) {
+			if (attempts > 0) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("executor(s) rejected message after " + attempts
+							+ " attempt(s), will try again after 'retryInterval' of " + this.policy.getRetryInterval()
+							+ " milliseconds");
+				}
+				try {
+					Thread.sleep(policy.getRetryInterval());
+				}
+				catch (InterruptedException iex) {
+					Thread.currentThread().interrupt();
+					return false;
+				}
 			}
-			return false;
-		}
-		while (iter.hasNext()) {
-			MessageReceivingExecutor executor = iter.next();
-			try {
+			Iterator<MessageReceivingExecutor> iter = this.getExecutors().iterator();
+			if (!iter.hasNext()) {
+				if (logger.isWarnEnabled()) {
+					logger.warn("dispatcher has no active executors");
+				}
+				return false;
+			}
+			while (iter.hasNext()) {
+				MessageReceivingExecutor executor = iter.next();
 				if (executor == null || !executor.isRunning()) {
 					if (logger.isInfoEnabled()) {
 						logger.info("skipping inactive executor");
 					}
 					continue;
 				}
-				executor.processMessage(message);
-				return true;
-			}
-			catch (RejectedExecutionException rex) {
-				attempts++;
-				if (attempts == policy.getRejectionLimit()) {
-					attempts = 0;
+				try {
+					executor.processMessage(message);
+					return true;
+				}
+				catch (RejectedExecutionException rex) {
 					if (logger.isDebugEnabled()) {
-						logger.debug("reached rejected execution limit");
-					}
-					try {
-						Thread.sleep(policy.getRejectionLimitWait());
-					}
-					catch (InterruptedException iex) {
-						Thread.currentThread().interrupt();
+						logger.debug("executor rejected task, continuing with other executors if available", rex);
 					}
 				}
 			}
-			catch (Exception e) {
-				if (logger.isWarnEnabled()) {
-					logger.warn("error occurred during dispatch", e);
-				}
-			}
+			attempts++;
 		}
-		return false;
+		throw new MessageDeliveryException("Dispatcher reached rejection limit of " +
+				this.policy.getRejectionLimit() + ". Consider increasing the concurrency and/or raising the limit.");
 	}
 
 }