GH-1487: @KafkaListener Meta-Annotation on Class

Resolves #1487

A meta-annotated annotation did not work at the class level.

**cherry-pick to 2.4.x, 2.3.x, 2.2.x**

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

@@ -313,7 +313,7 @@ public Object postProcessAfterInitialization(final Object bean, final String bea
 	 */
 	private Collection<KafkaListener> findListenerAnnotations(Class<?> clazz) {
 		Set<KafkaListener> listeners = new HashSet<>();
-		KafkaListener ann = AnnotationUtils.findAnnotation(clazz, KafkaListener.class);
+		KafkaListener ann = AnnotatedElementUtils.findMergedAnnotation(clazz, KafkaListener.class);
 		if (ann != null) {
 			listeners.add(ann);
 		}

spring-kafka/src/test/java/org/springframework/kafka/annotation/AliasPropertiesTests.java

@@ -33,6 +33,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.annotation.AliasFor;
+import org.springframework.kafka.annotation.AliasPropertiesTests.Config.ClassLevelListener;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
@@ -43,6 +44,7 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
@@ -66,11 +68,15 @@ public class AliasPropertiesTests {
 	@Autowired
 	private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
 
+	@Autowired
+	private ClassLevelListener classLevel;
+
 	@Test
 	public void testAliasFor() throws Exception {
 		this.template.send("alias.tests", "foo");
 		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.kafkaListenerEndpointRegistry()).isSameAs(this.kafkaListenerEndpointRegistry);
+		assertThat(this.classLevel.latch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
 	@Configuration
@@ -125,18 +131,34 @@ public Map<String, Object> producerConfigs() {
 			return KafkaTestUtils.producerProps(embeddedKafka());
 		}
 
-		@MyListener("alias.tests")
+		@MyListener(id = "onMethod", value = "alias.tests")
 		public void listen1(String in) {
-			latch.countDown();
+			this.latch.countDown();
+		}
+
+		@Component
+		@MyListener(id = "onClass", value = "alias.tests")
+		public static class ClassLevelListener {
+
+			private final CountDownLatch latch = new CountDownLatch(1);
+
+			@KafkaHandler
+			void listen(String in) {
+				this.latch.countDown();
+			}
+
 		}
 
 	}
 
-	@Target(ElementType.METHOD)
+	@Target({ ElementType.METHOD, ElementType.TYPE })
 	@Retention(RetentionPolicy.RUNTIME)
 	@KafkaListener
 	public @interface MyListener {
 
+		@AliasFor(annotation = KafkaListener.class, attribute = "id")
+		String id() default "";
+
 		@AliasFor(annotation = KafkaListener.class, attribute = "topics")
 		String[] value();