RetryTemplate per binding

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java

@@ -180,6 +180,10 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 	 */
 	protected RetryTemplate buildRetryTemplate(ConsumerProperties properties) {
 		RetryTemplate rt = this.consumerBindingRetryTemplate;
+		rt = properties.getRetryTemplate() == null ?
+        rt :
+        getBeanFactory().getBean(properties.getRetryTemplate(), RetryTemplate.class);
+
 		if (rt == null) {
 			rt = new RetryTemplate();
 			SimpleRetryPolicy retryPolicy = CollectionUtils.isEmpty(properties.getRetryableExceptions())

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java

@@ -80,6 +80,8 @@ public class ConsumerProperties {
 	 * Default: 3. Set to 1 to disable retry. You can also provide custom RetryTemplate
 	 * in the event you want to take complete control of the RetryTemplate. Simply configure
 	 * it as @Bean inside your application configuration.
+	 * If you need to be binding specific, then you can reference a specific RetryTemplate by name
+	 * with the retry-template configuration.
 	 */
 	private int maxAttempts = 3;
 
@@ -119,6 +121,13 @@ public class ConsumerProperties {
 	 */
 	private boolean defaultRetryable = true;
 
+	/**
+	 * Instead of relying on the framework default or global RetryTemplate specified by
+	 * StreamRetryTemplate, you can provide the framework a bean by name to look up in the
+	 * ApplicationContext.
+	 */
+	private String retryTemplate = null;
+
 	/**
 	 * A map of Throwable class names in the key and a boolean in the value.
 	 * Specify those exceptions (and subclasses) that will or won't be retried.
@@ -158,6 +167,14 @@ public class ConsumerProperties {
 	 */
 	private boolean multiplex;
 
+	public String getRetryTemplate(){
+		return retryTemplate;
+	}
+
+	public void setRetryTemplate(String retryTemplate){
+		this.retryTemplate = retryTemplate;
+	}
+
 	@Min(value = 1, message = "Concurrency should be greater than zero.")
 	public int getConcurrency() {
 		return concurrency;