GH-2441: Merged Annotations with RetryableTopic

Resolves #2441

Add merged annotations support to RetryableTopic

Add merged attribute to meta-annotation example

Add https to license url

Docs.

Author: fablakeson

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

@@ -82,6 +82,23 @@ public void processMessage(MyPojo message) {
 NOTE: If you don't specify a kafkaTemplate name a bean with name `defaultRetryTopicKafkaTemplate` will be looked up.
 If no bean is found an exception is thrown.
 
+Starting with version 3.0, the `@RetryableTopic` annotation can be used as a meta-annotation on custom annotations; for example:
+
+====
+[source, java]
+----
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@RetryableTopic
+static @interface MetaAnnotatedRetryableTopic {
+
+    @AliasFor(attribute = "concurrency", annotation = RetryableTopic.class)
+    String parallelism() default "3";
+
+}
+----
+====
+
 ===== Using `RetryTopicConfiguration` beans
 
 You can also configure the non-blocking retry support by creating `RetryTopicConfiguration` beans in a `@Configuration` annotated class.

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

@@ -38,6 +38,8 @@ The bootstrapping of <<retry-topic>> infrastructure beans has changed in this re
 
 You can now set a different `concurrency` for the retry containers; by default, the concurrency is the same as the main container.
 
+`@RetryableTopic` can now be used as a meta-annotation on custom annotations, including support for `@AliasFor` properties.
+
 See <<retry-config>> for more information.
 
 You can now configure multiple `@RetryableTopic` listeners on the same topic in the same application context.

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

@@ -27,7 +27,10 @@
 import org.springframework.beans.factory.config.BeanExpressionResolver;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.context.expression.StandardBeanExpressionResolver;
-import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.annotation.MergedAnnotation;
+import org.springframework.core.annotation.MergedAnnotations;
+import org.springframework.core.annotation.MergedAnnotations.SearchStrategy;
+import org.springframework.core.annotation.RepeatableContainers;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
 
@@ -88,7 +91,11 @@ public RetryTopicConfigurationProvider(BeanFactory beanFactory, BeanExpressionRe
 		this.expressionContext = expressionContext;
 	}
 	public RetryTopicConfiguration findRetryConfigurationFor(String[] topics, Method method, Object bean) {
-		RetryableTopic annotation = AnnotationUtils.findAnnotation(method, RetryableTopic.class);
+		RetryableTopic annotation = MergedAnnotations.from(method, SearchStrategy.TYPE_HIERARCHY,
+					RepeatableContainers.none())
+				.get(RetryableTopic.class)
+				.synthesize(MergedAnnotation::isPresent)
+				.orElse(null);
 		return annotation != null
 				? new RetryableTopicAnnotationProcessor(this.beanFactory, this.resolver, this.expressionContext)
 						.processAnnotation(topics, method, annotation, bean)

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

@@ -150,11 +150,13 @@
  *</pre>
  * <p> Or through meta-annotations, such as:
  * <pre>
- *     <code>@RetryableTopic(attempts = 3,
- *     		backoff = @Backoff(delay = 700, maxDelay = 12000, multiplier = 3))</code>
- *     <code>public @interface WithExponentialBackoffRetry { }</code>
+ *     <code>@RetryableTopic(backoff = @Backoff(delay = 700, maxDelay = 12000, multiplier = 3))</code>
+ *     <code>public @interface WithExponentialBackoffRetry {</code>
+ *     <code>   	{@literal @}AliasFor(attribute = "attempts", annotation = RetryableTopic.class)
+ *        	String retries();
+ *     }</code>
  *
- *     <code>@WithExponentialBackoffRetry</code>
+ *     <code>@WithExponentialBackoffRetry(retries = "3")</code>
  *     <code>@KafkaListener(topics = "my-annotated-topic")
  *     public void processMessage(MyPojo message) {
  *        		// ... message processing

spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingCoverageTests.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ *      https://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationProviderTests.java

@@ -38,6 +38,7 @@
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.core.annotation.AliasFor;
 import org.springframework.kafka.annotation.RetryTopicConfigurationProvider;
 import org.springframework.kafka.annotation.RetryableTopic;
 import org.springframework.kafka.core.KafkaOperations;
@@ -155,6 +156,8 @@ void shouldProvideFromMetaAnnotation() {
 
 		// then
 		then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class);
+		assertThat(configuration.getConcurrency()).isEqualTo(3);
+
 	}
 
 	@Test
@@ -182,6 +185,8 @@ public void nonAnnotatedMethod() {
 	@Retention(RetentionPolicy.RUNTIME)
 	@RetryableTopic
 	static @interface MetaAnnotatedRetryableTopic {
+		@AliasFor(attribute = "concurrency", annotation = RetryableTopic.class)
+		String parallelism() default "3";
 	}
 
 	@MetaAnnotatedRetryableTopic