Add meta-annotation support to RetryableTopic

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.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.
@@ -37,11 +37,12 @@
  *
  * @author Tomaz Fernandes
  * @author Gary Russell
+ * @author Fabio da Silva Jr.
  * @since 2.7
  *
  * @see org.springframework.kafka.retrytopic.RetryTopicConfigurer
  */
-@Target({ ElementType.METHOD })
+@Target({ ElementType.METHOD, ElementType.ANNOTATION_TYPE })
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 public @interface RetryableTopic {

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

@@ -138,7 +138,7 @@
  *
  * <p>The other, non-exclusive way to configure the endpoints is through the convenient
  * {@link org.springframework.kafka.annotation.RetryableTopic} annotation, that can be placed on any
- * {@link org.springframework.kafka.annotation.KafkaListener} annotated methods, such as:
+ * {@link org.springframework.kafka.annotation.KafkaListener} annotated methods, directly, such as:
  *
  * <pre>
  *     <code>@RetryableTopic(attempts = 3,
@@ -148,6 +148,18 @@
  *        		// ... message processing
  *     }</code>
  *</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>@WithExponentialBackoffRetry</code>
+ *     <code>@KafkaListener(topics = "my-annotated-topic")
+ *     public void processMessage(MyPojo message) {
+ *        		// ... message processing
+ *     }</code>
+ *</pre>
  * <p> The same configurations are available in the annotation and the builder approaches, and both can be
  * used concurrently. In case the same method / topic can be handled by both, the annotation takes precedence.
  *
@@ -192,6 +204,7 @@
  * If no DLT handler is provided, the default {@link LoggingDltListenerHandlerMethod} is used.
  *
  * @author Tomaz Fernandes
+ * @author Fabio da Silva Jr.
  * @since 2.7
  *
  * @see RetryTopicConfigurationBuilder

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

@@ -25,6 +25,10 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
 import java.lang.reflect.Method;
 import java.util.Collections;
 
@@ -41,6 +45,7 @@
 /**
  * @author Tomaz Fernandes
  * @author Gary Russell
+ * @author Fabio da Silva Jr.
  * @since 2.7
  */
 @ExtendWith(MockitoExtension.class)
@@ -61,6 +66,8 @@ class RetryTopicConfigurationProviderTests {
 
 	private final Method nonAnnotatedMethod = getAnnotatedMethod("nonAnnotatedMethod");
 
+	private final Method metaAnnotatedMethod = getAnnotatedMethod("metaAnnotatedMethod");
+
 	private Method getAnnotatedMethod(String methodName) {
 		try {
 			return  this.getClass().getDeclaredMethod(methodName);
@@ -136,6 +143,19 @@ void shouldFindNone() {
 
 	}
 
+	@Test
+	void shouldProvideFromMetaAnnotation() {
+
+		// setup
+		willReturn(kafkaOperations).given(beanFactory).getBean("retryTopicDefaultKafkaTemplate", KafkaOperations.class);
+
+		// given
+		RetryTopicConfigurationProvider provider = new RetryTopicConfigurationProvider(beanFactory);
+		RetryTopicConfiguration configuration = provider.findRetryConfigurationFor(topics, metaAnnotatedMethod, bean);
+
+		// then
+		then(this.beanFactory).should(times(0)).getBeansOfType(RetryTopicConfiguration.class);
+	}
 
 	@Test
 	void shouldNotConfigureIfBeanFactoryNull() {
@@ -157,4 +177,15 @@ public void annotatedMethod() {
 	public void nonAnnotatedMethod() {
 		// NoOps
 	}
+
+	@Target({ElementType.METHOD})
+	@Retention(RetentionPolicy.RUNTIME)
+	@RetryableTopic
+	static @interface MetaAnnotatedRetryableTopic {
+	}
+
+	@MetaAnnotatedRetryableTopic
+	public void metaAnnotatedMethod() {
+		// NoOps
+	}
 }