GH-1460: Configurable Exceptions for Retry

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1460

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

@@ -977,6 +977,17 @@ backOffMultiplier::
 The backoff multiplier.
 +
 Default 2.0.
+defaultRetryable::
+Whether exceptions thrown by the listener that are not listed in the `retryableExceptions` are retryable.
++
+Default: `true`.
+retryableExceptions::
+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.
+Also see `defaultRetriable`.
+Example: `spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false`.
++
+Default: empty.
 
 While the preceding settings are sufficient for majority of the customization requirements, they may not satisfy certain complex requirements at, which
 point you may want to provide your own instance of the `RetryTemplate`. To do so configure it as a bean in your application configuration. The application provided
@@ -1507,6 +1518,10 @@ backOffMultiplier::
 The backoff multiplier.
 +
 Default: `2.0`.
+defaultRetryable::
+Whether exceptions thrown by the listener that are not listed in the `retryableExceptions` are retryable.
++
+Default: `true`.
 instanceIndex::
 When set to a value greater than equal to zero, it allows customizing the instance index of this consumer (if different from `spring.cloud.stream.instanceIndex`).
 When set to a negative value, it defaults to `spring.cloud.stream.instanceIndex`.
@@ -1519,6 +1534,13 @@ When set to a negative value, it defaults to `spring.cloud.stream.instanceCount`
 See "`<<spring-cloud-stream-overview-instance-index-instance-count>>`" for more information.
 +
 Default: `-1`.
+retryableExceptions::
+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.
+Also see `defaultRetriable`.
+Example: `spring.cloud.stream.bindings.input.consumer.retryable-exceptions.java.lang.IllegalStateException=false`.
++
+Default: empty.
 useNativeDecoding::
 When set to `true`, the inbound message is deserialized directly by the client library, which must be configured correspondingly (for example, setting an appropriate Kafka producer value deserializer).
 When this configuration is being used, the inbound message unmarshalling is not based on the `contentType` of the binding.

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

@@ -180,8 +180,14 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
 		RetryTemplate rt = this.consumerBindingRetryTemplate;
 		if (rt == null) {
 			rt = new RetryTemplate();
-			SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
-			retryPolicy.setMaxAttempts(properties.getMaxAttempts());
+			SimpleRetryPolicy retryPolicy;
+			if (properties.getRetryableExceptions().size() == 0) {
+				retryPolicy = new SimpleRetryPolicy(properties.getMaxAttempts());
+			}
+			else {
+				retryPolicy = new SimpleRetryPolicy(properties.getMaxAttempts(), properties.getRetryableExceptions(),
+						true, properties.isDefaultRetryable());
+			}
 			ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
 			backOffPolicy.setInitialInterval(properties.getBackOffInitialInterval());
 			backOffPolicy.setMultiplier(properties.getBackOffMultiplier());

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

@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.binder;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
+
 import javax.validation.constraints.Min;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
@@ -104,6 +107,18 @@ public class ConsumerProperties implements MergableProperties{
 	 */
 	private double backOffMultiplier = 2.0;
 
+	/**
+	 * Whether exceptions thrown by the listener that are not listed in the
+	 * 'retryableExceptions' are retryable.
+	 */
+	private boolean defaultRetryable = true;
+
+	/**
+	 * 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.
+	 */
+	private Map<Class<? extends Throwable>, Boolean> retryableExceptions = new LinkedHashMap<>();
+
 	/**
 	 * When set to none, disables header parsing on input. Effective only
 	 * for messaging middleware that does not support message headers natively
@@ -208,6 +223,22 @@ public class ConsumerProperties implements MergableProperties{
 		this.backOffMultiplier = backOffMultiplier;
 	}
 
+	public boolean isDefaultRetryable() {
+		return this.defaultRetryable;
+	}
+
+	public void setDefaultRetryable(boolean defaultRetryable) {
+		this.defaultRetryable = defaultRetryable;
+	}
+
+	public Map<Class<? extends Throwable>, Boolean> getRetryableExceptions() {
+		return this.retryableExceptions;
+	}
+
+	public void setRetryableExceptions(Map<Class<? extends Throwable>, Boolean> retryableExceptions) {
+		this.retryableExceptions = retryableExceptions;
+	}
+
 	public HeaderMode getHeaderMode() {
 		return this.headerMode;
 	}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java

@@ -271,6 +271,7 @@ public class PollableConsumerTests {
 		ExtendedConsumerProperties<Object> properties = new ExtendedConsumerProperties<>(null);
 		properties.setMaxAttempts(2);
 		properties.setBackOffInitialInterval(0);
+		properties.getRetryableExceptions().put(IllegalStateException.class, false);
 		binder.bindPollableConsumer("foo", "bar", pollableSource, properties);
 		final CountDownLatch latch = new CountDownLatch(1);
 		context.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, SubscribableChannel.class).subscribe(m -> {
@@ -285,6 +286,14 @@ public class PollableConsumerTests {
 		Message<?> lastError = binder.getLastError();
 		assertThat(lastError).isNotNull();
 		assertThat(((Exception) lastError.getPayload()).getCause().getMessage()).isEqualTo("test recoverer");
+		assertThat(pollableSource.poll(received -> {
+			count.incrementAndGet();
+			throw new IllegalStateException("no retries");
+		})).isTrue();
+		assertThat(count.get()).isEqualTo(3);
+		lastError = binder.getLastError();
+		assertThat(lastError).isNotNull();
+		assertThat(((Exception) lastError.getPayload()).getCause().getMessage()).isEqualTo("no retries");
 	}
 
 	@Test