GH-1511 Polishing and added tests

- Polished RetryTemplate changes from the previous commit
- Rebased with 2.2.x branch
- Added tests
- Updated author tags with original committer's name

Resolves #1511
Resolves #1588

docs/src/main/asciidoc/spring-cloud-stream.adoc

@@ -1054,6 +1054,15 @@ public RetryTemplate myRetryTemplate() {
 ----
 As you can see from the above example you don't need to annotate it with `@Bean` since `@StreamRetryTemplate` is a qualified `@Bean`.
 
+If you need to be more precise with your `RetryTemplate`, you can specify the bean by name in your `ConsumerProperties` to associate
+the specific retry bean per binding.
+
+[source]
+----
+spring.cloud.stream.bindings.<foo>.consumer.retry-template-name=<your-retry-template-bean-name>
+----
+
+
 [[spring-cloud-stream-overview-binders]]
 == Binders
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2018 the original author or authors.
+ * Copyright 2013-2019 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder;
 
+import java.util.Map;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -49,6 +51,7 @@ import org.springframework.util.StringUtils;
  * @author Soby Chacko
  * @author Vinicius Carvalho
  * @author Oleg Zhurakousky
+ * @author Nicolas Homble
  */
 public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends ProducerProperties>
 	implements ApplicationContextAware, InitializingBean, Binder<T, C, P> {
@@ -65,9 +68,9 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 
 	private volatile EvaluationContext evaluationContext;
 
-	@Autowired(required = false) // this would need to be refactored into constructor in the future
+	@Autowired(required = false)
 	@StreamRetryTemplate
-	private RetryTemplate consumerBindingRetryTemplate;
+	private Map<String, RetryTemplate> consumerBindingRetryTemplates;
 
 	/**
 	 * For binder implementations that support a prefix, apply the prefix to the name.
@@ -179,12 +182,8 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 	 * @return The retry template
 	 */
 	protected RetryTemplate buildRetryTemplate(ConsumerProperties properties) {
-		RetryTemplate rt = this.consumerBindingRetryTemplate;
-		rt = properties.getRetryTemplate() == null ?
-        rt :
-        getBeanFactory().getBean(properties.getRetryTemplate(), RetryTemplate.class);
-
-		if (rt == null) {
+		RetryTemplate rt;
+		if (CollectionUtils.isEmpty(this.consumerBindingRetryTemplates)) {
 			rt = new RetryTemplate();
 			SimpleRetryPolicy retryPolicy = CollectionUtils.isEmpty(properties.getRetryableExceptions())
 					? new SimpleRetryPolicy(properties.getMaxAttempts())
@@ -197,6 +196,11 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 			rt.setRetryPolicy(retryPolicy);
 			rt.setBackOffPolicy(backOffPolicy);
 		}
+		else {
+			rt = StringUtils.hasText(properties.getRetryTemplateName())
+					? this.consumerBindingRetryTemplates.get(properties.getRetryTemplateName())
+							: this.consumerBindingRetryTemplates.values().iterator().next();
+		}
 		return rt;
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -25,13 +25,14 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 
 
 /**
- * Common consumer properties.
+ * Common consumer properties - spring.cloud.stream.bindings.[destinationName].consumer.
  *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
  * @author Gary Russell
  * @author Soby Chacko
  * @author Oleg Zhurakousky
+ * @author Nicolas Homble
  */
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
@@ -122,11 +123,9 @@ 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.
+	 * Allows you to further qualify which RetryTemplate to use for a specific consumer binding..
 	 */
-	private String retryTemplate = null;
+	private String retryTemplateName;
 
 	/**
 	 * A map of Throwable class names in the key and a boolean in the value.
@@ -167,12 +166,12 @@ public class ConsumerProperties {
 	 */
 	private boolean multiplex;
 
-	public String getRetryTemplate(){
-		return retryTemplate;
+	public String getRetryTemplateName(){
+		return retryTemplateName;
 	}
 
-	public void setRetryTemplate(String retryTemplate){
-		this.retryTemplate = retryTemplate;
+	public void setRetryTemplateName(String retryTemplateName){
+		this.retryTemplateName = retryTemplateName;
 	}
 
 	@Min(value = 1, message = "Concurrency should be greater than zero.")

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java

@@ -0,0 +1,119 @@
+/*
+ * Copyright 2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.springframework.cloud.stream.config;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Map;
+
+import org.junit.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamRetryTemplate;
+import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.retry.support.RetryTemplate;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class RetryTemplateTests {
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testSingleCustomRetryTemplate() throws Exception {
+		ApplicationContext context = new SpringApplicationBuilder(SingleCustomRetryTemplateConfiguration.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		AbstractBinder binder = context.getBean(AbstractBinder.class);
+		Field f = AbstractBinder.class.getDeclaredField("consumerBindingRetryTemplates");
+		f.setAccessible(true);
+		@SuppressWarnings("unchecked")
+		Map<String, RetryTemplate> consumerBindingRetryTemplates = (Map<String, RetryTemplate>) f.get(binder);
+		assertTrue(consumerBindingRetryTemplates.size() == 1);
+	}
+
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class SingleCustomRetryTemplateConfiguration {
+
+		@StreamRetryTemplate
+		public RetryTemplate retryTemplate() {
+			return new RetryTemplate();
+		}
+
+		@Bean
+		public RetryTemplate otherRetryTemplate() {
+			return new RetryTemplate();
+		}
+	}
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	public void testSpecificCustomRetryTemplate() throws Exception {
+		ApplicationContext context = new SpringApplicationBuilder(SpecificCustomRetryTemplateConfiguration.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false",
+						"--spring.cloud.stream.bindings.input.consumer.retry-template-name=retryTemplateTwo");
+
+		RetryTemplate retryTemplateTwo = context.getBean("retryTemplateTwo", RetryTemplate.class);
+		BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class);
+		ConsumerProperties consumerProperties = bindingServiceProperties.getConsumerProperties("input");
+		AbstractBinder binder = context.getBean(AbstractBinder.class);
+
+		Method m = AbstractBinder.class.getDeclaredMethod("buildRetryTemplate", ConsumerProperties.class);
+		m.setAccessible(true);
+		RetryTemplate retryTemplate = (RetryTemplate) m.invoke(binder, consumerProperties);
+		assertEquals(retryTemplateTwo, retryTemplate);
+	}
+
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class SpecificCustomRetryTemplateConfiguration {
+
+		@StreamRetryTemplate
+		public RetryTemplate retryTemplate() {
+			return new RetryTemplate();
+		}
+
+		@StreamRetryTemplate
+		public RetryTemplate retryTemplateTwo() {
+			return new RetryTemplate();
+		}
+
+		@Bean
+		public RetryTemplate otherRetryTemplate() {
+			return new RetryTemplate();
+		}
+	}
+}