GH-2443: Distinct Concurrency for Retry Containers

Resolves #2443

Optionally configure a different concurrency for retry containers.

* Add Javadocs to RetryTopicConfigurationBuilder.

* Javadoc polishing.

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/retrytopic.adoc

@@ -46,6 +46,9 @@ NOTE: It is not necessary to also add `@EnableKafka`, if you add this annotation
 Also, starting with that version, for more advanced configuration of the feature's components and global features, the `RetryTopicConfigurationSupport` class should be extended in a `@Configuration` class, and the appropriate methods overridden.
 For more details refer to <<retry-topic-global-settings>>.
 
+By default, the containers for the retry topics will have the same concurrency as the main container.
+Starting with version 3.0, you can set a different `concurrency` for the retry containers (either on the annotation, or in `RetryConfigurationBuilder`).
+
 IMPORTANT: Only one of the above techniques can be used, and only one `@Configuration` class can extend `RetryTopicConfigurationSupport`.
 
 ===== Using the `@RetryableTopic` annotation
@@ -108,6 +111,7 @@ public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> templa
             .newInstance()
             .fixedBackoff(3000)
             .maxAttempts(5)
+            .concurrency(1)
             .includeTopics("my-topic", "my-other-topic")
             .create(template);
 }

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -35,6 +35,9 @@ See <<same-broker-multiple-tests>> for more information.
 This feature is no longer considered experimental (as far as its API is concerned), the feature itself has been supported since 2.7, but with a greater than normal possibility of breaking API changes.
 
 The bootstrapping of <<retry-topic>> infrastructure beans has changed in this release to avoid some timing problems that occurred in some application regarding application initialization.
+
+You can now set a different `concurrency` for the retry containers; by default, the concurrency is the same as the main container.
+
 See <<retry-config>> for more information.
 
 [[x30-lc-changes]]

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java

@@ -194,4 +194,12 @@
 	 */
 	String autoStartDltHandler() default "";
 
+	/**
+	 * Concurrency for the retry and DLT containers; if not specified, the main container
+	 * concurrency is used.
+	 * @return the concurrency.
+	 * @since 3.0
+	 */
+	String concurrency() default "";
+
 }

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopicAnnotationProcessor.java

@@ -129,6 +129,7 @@ public RetryTopicConfiguration processAnnotation(String[] topics, Method method,
 		}
 		return RetryTopicConfigurationBuilder.newInstance()
 				.maxAttempts(resolveExpressionAsInteger(annotation.attempts(), "attempts", true))
+				.concurrency(resolveExpressionAsInteger(annotation.concurrency(), "concurrency", false))
 				.customBackoff(createBackoffFromAnnotation(annotation.backoff(), this.beanFactory))
 				.retryTopicSuffix(resolveExpressionAsString(annotation.retryTopicSuffix(), "retryTopicSuffix"))
 				.dltSuffix(resolveExpressionAsString(annotation.dltTopicSuffix(), "dltTopicSuffix"))

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -20,6 +20,7 @@
 
 import org.springframework.kafka.support.AllowDenyCollectionManager;
 import org.springframework.kafka.support.EndpointHandlerMethod;
+import org.springframework.lang.Nullable;
 
 /**
  * Contains the provided configuration for the retryable topics.
@@ -45,18 +46,23 @@ public class RetryTopicConfiguration {
 
 	private final ListenerContainerFactoryConfigurer.Configuration factoryConfigurerConfig;
 
+	@Nullable
+	private final Integer concurrency;
+
 	RetryTopicConfiguration(List<DestinationTopic.Properties> destinationTopicProperties,
 							EndpointHandlerMethod dltHandlerMethod,
 							TopicCreation kafkaTopicAutoCreationConfig,
 							AllowDenyCollectionManager<String> topicAllowListManager,
 							ListenerContainerFactoryResolver.Configuration factoryResolverConfig,
-							ListenerContainerFactoryConfigurer.Configuration factoryConfigurerConfig) {
+							ListenerContainerFactoryConfigurer.Configuration factoryConfigurerConfig,
+							@Nullable Integer concurrency) {
 		this.destinationTopicProperties = destinationTopicProperties;
 		this.dltHandlerMethod = dltHandlerMethod;
 		this.kafkaTopicAutoCreationConfig = kafkaTopicAutoCreationConfig;
 		this.topicAllowListManager = topicAllowListManager;
 		this.factoryResolverConfig = factoryResolverConfig;
 		this.factoryConfigurerConfig = factoryConfigurerConfig;
+		this.concurrency = concurrency;
 	}
 
 	public boolean hasConfigurationForTopics(String[] topics) {
@@ -83,6 +89,11 @@ public List<DestinationTopic.Properties> getDestinationTopicProperties() {
 		return this.destinationTopicProperties;
 	}
 
+	@Nullable
+	public Integer getConcurrency() {
+		return this.concurrency;
+	}
+
 	static class TopicCreation {
 
 		private final boolean shouldCreateTopics;