ConcurrentHandler and DefaultMessageEndpoint now propagate rejection exceptions so that DispatcherTask has proper retry behavior, and PollingSchedule now implements equals and hashCode so that the handler map keys are recognized (handlers are added to the same task if they have the same schedule).

spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java

@@ -29,6 +29,7 @@ import org.springframework.integration.channel.ChannelRegistry;
 import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.MessageHandlerNotRunningException;
+import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
 import org.springframework.integration.dispatcher.MessageSelectorRejectedException;
 import org.springframework.integration.handler.ConcurrentHandler;
 import org.springframework.integration.handler.MessageHandler;
@@ -216,6 +217,9 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
 				replyChannel.send(replyMessage);
 			}
 		}
+		catch (MessageHandlerRejectedExecutionException e) {
+			throw e;
+		}
 		catch (Throwable t) {
 			if (this.errorHandler == null) {
 				throw new MessageHandlingException(

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

@@ -16,8 +16,6 @@
 
 package org.springframework.integration.handler;
 
-import java.util.concurrent.RejectedExecutionException;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -164,7 +162,7 @@ public class ConcurrentHandler implements MessageHandler, Lifecycle, Initializin
 			this.executor.execute(new HandlerTask(message));
 			return null;
 		}
-		catch (RejectedExecutionException e) {
+		catch (RuntimeException e) {
 			throw new MessageHandlerRejectedExecutionException(e);
 		}
 	}

spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java

@@ -83,4 +83,21 @@ public class PollingSchedule implements Schedule {
 		this.fixedRate = fixedRate;
 	}
 
+	public int hashCode() {
+		return toString().hashCode() * 23;
+	}
+
+	public boolean equals(Object other) {
+		if (other == null) {
+			return false;
+		}
+		return (other instanceof PollingSchedule && 
+				this.toString().equals(((PollingSchedule) other).toString()));
+	}
+
+	public String toString() {
+		return "initialDelay=" + this.initialDelay + ", period=" + this.period +
+				", timeUnit=" + this.timeUnit + ", fixedRate=" + this.fixedRate;
+	}
+
 }