GH-830: Enable usage of authorizationExceptionRetryInterval

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/830

Enable binder configuration of authorizationExceptionRetryInterval through properties

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
 

pom.xml

@@ -12,7 +12,7 @@
 	</parent>
 	<properties>
 		<java.version>1.8</java.version>
-		<spring-kafka.version>2.3.2.RELEASE</spring-kafka.version>
+		<spring-kafka.version>2.3.5.RELEASE</spring-kafka.version>
 		<spring-integration-kafka.version>3.2.1.RELEASE</spring-integration-kafka.version>
 		<kafka.version>2.3.1</kafka.version>
 		<spring-cloud-schema-registry.version>1.0.2.BUILD-SNAPSHOT</spring-cloud-schema-registry.version>

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.
 	 */

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) {