diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 0e8376509..ed2e7111a 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/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 "`<>`" 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. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index 22a09e531..c13b5039b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -180,8 +180,14 @@ public abstract class AbstractBinder, 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, Boolean> getRetryableExceptions() { + return this.retryableExceptions; + } + + public void setRetryableExceptions(Map, Boolean> retryableExceptions) { + this.retryableExceptions = retryableExceptions; + } + public HeaderMode getHeaderMode() { return this.headerMode; } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java index 42ffa3aa8..a6882bd98 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/PollableConsumerTests.java @@ -271,6 +271,7 @@ public class PollableConsumerTests { ExtendedConsumerProperties 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