diff --git a/README.adoc b/README.adoc index 45d973c74..0792be38e 100644 --- a/README.adoc +++ b/README.adoc @@ -159,6 +159,13 @@ Use this, for example, if you wish to customize the trusted packages in a `Defau + Default: none. +spring.cloud.stream.kafka.binder.authorizationExceptionRetryInterval:: +Enables retrying in case of authorization exceptions. +Defines interval between each retry. +Accepts `Duration`, e.g. `30s`, `2m`, etc. ++ +Default: `null` (retries disabled, fail fast) + [[kafka-consumer-properties]] ==== Kafka Consumer Properties diff --git a/pom.xml b/pom.xml index 274ee06e4..7b8fdee6e 100644 --- a/pom.xml +++ b/pom.xml @@ -12,7 +12,7 @@ 1.8 - 2.3.2.RELEASE + 2.3.5.RELEASE 3.2.1.RELEASE 2.3.1 1.0.2.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java index 3488a74d6..704f46e05 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.properties; +import java.time.Duration; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -51,6 +52,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Rafal Zukowski * @author Aldo Sinanaj + * @author Lukasz Kaminski */ @ConfigurationProperties(prefix = "spring.cloud.stream.kafka.binder") public class KafkaBinderConfigurationProperties { @@ -107,6 +109,13 @@ public class KafkaBinderConfigurationProperties { */ private String headerMapperBeanName; + /** + * Time between retries after AuthorizationException is caught in + * the ListenerContainer; defalt is null which disables retries. + * For more info see: {@link org.springframework.kafka.listener.ConsumerProperties#setAuthorizationExceptionRetryInterval(java.time.Duration)} + */ + private Duration authorizationExceptionRetryInterval; + public KafkaBinderConfigurationProperties(KafkaProperties kafkaProperties) { Assert.notNull(kafkaProperties, "'kafkaProperties' cannot be null"); this.kafkaProperties = kafkaProperties; @@ -346,6 +355,14 @@ public class KafkaBinderConfigurationProperties { this.headerMapperBeanName = headerMapperBeanName; } + public Duration getAuthorizationExceptionRetryInterval() { + return authorizationExceptionRetryInterval; + } + + public void setAuthorizationExceptionRetryInterval(Duration authorizationExceptionRetryInterval) { + this.authorizationExceptionRetryInterval = authorizationExceptionRetryInterval; + } + /** * Domain class that models transaction capabilities in Kafka. */ diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 08e7526c5..6f48f8adb 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -143,6 +143,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback; * @author Soby Chacko * @author Henryk Konsek * @author Doug Saus + * @author Lukasz Kaminski */ public class KafkaMessageChannelBinder extends // @checkstyle:off @@ -594,6 +595,7 @@ public class KafkaMessageChannelBinder extends concurrency = extendedConsumerProperties.getConcurrency(); } resetOffsetsForAutoRebalance(extendedConsumerProperties, consumerFactory, containerProperties); + containerProperties.setAuthorizationExceptionRetryInterval(this.configurationProperties.getAuthorizationExceptionRetryInterval()); @SuppressWarnings("rawtypes") final ConcurrentMessageListenerContainer messageListenerContainer = new ConcurrentMessageListenerContainer( consumerFactory, containerProperties) {