Addressing PR review

Signed-off-by: chickenchickenlove <[email protected]>

Author: chickenchickenlove

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

@@ -20,6 +20,7 @@
 import java.io.StringReader;
 import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
@@ -33,7 +34,6 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -149,7 +149,6 @@
  * @author Sanghyeok An
  * @author Soby Chacko
  * @author Omer Celik
- * @author Sanghyeok An
  *
  * @see KafkaListener
  * @see KafkaListenerErrorHandler
@@ -159,7 +158,6 @@
  * @see KafkaListenerEndpointRegistry
  * @see org.springframework.kafka.config.KafkaListenerEndpoint
  * @see MethodKafkaListenerEndpoint
- * @see KafkaHandlerParser
  */
 public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 		implements BeanPostProcessor, Ordered, ApplicationContextAware, SmartInitializingSingleton {
@@ -216,8 +214,6 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	private final Lock globalLock = new ReentrantLock();
 
-	private final KafkaHandlerParser kafkaHandlerParser = new KafkaHandlerParser();
-
 	@Override
 	public int getOrder() {
 		return LOWEST_PRECEDENCE;
@@ -413,16 +409,23 @@ public Object postProcessAfterInitialization(final Object bean, final String bea
 							+ beanName + "': " + annotatedMethods);
 				}
 				if (hasClassLevelListeners) {
-					Optional<Method> methodWithoutAnnotation = this.kafkaHandlerParser.parseSingleHandlerMethod(targetClass);
-					if (methodWithoutAnnotation.isPresent() && classLevelListeners.size() == 1) {
+					Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass,
+							(ReflectionUtils.MethodFilter) method ->
+									AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
+					Set<Method> publicMethods = MethodIntrospector.selectMethods(targetClass,
+							(ReflectionUtils.MethodFilter) method ->
+									Modifier.isPublic(method.getModifiers()));
+
+					if (methodsWithHandler.isEmpty() && publicMethods.size() == 1 && !hasMethodLevelListeners) {
 						// Case when target class has class-level @KafkaListener annotation and
 						// has only single public method without @KafkaHandler.
-						for (KafkaListener kafkaListener : classLevelListeners) {
-							processKafkaListener(kafkaListener, methodWithoutAnnotation.get(), bean, beanName);
+						// See, GH-3807 for more details.
+						Method method = publicMethods.iterator().next();
+						for (KafkaListener classLevelListener : classLevelListeners) {
+							processKafkaListener(classLevelListener, method, bean, beanName);
 						}
 					}
 					else {
-						Set<Method> methodsWithHandler = MethodIntrospector.selectMethods(targetClass, (ReflectionUtils.MethodFilter) method -> AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
 						List<Method> multiMethods = new ArrayList<>(methodsWithHandler);
 						processMultiMethodListeners(classLevelListeners, multiMethods, targetClass, bean, beanName);
 					}

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

@@ -31,15 +31,14 @@
 import org.springframework.kafka.annotation.KafkaListenerTests.Config.ClassLevelListenerWithSinglePublicMethodAndPrivateMethod;
 import org.springframework.kafka.annotation.KafkaListenerTests.Config.OtherClassLevelListenerWithSinglePublicMethod;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
-import org.springframework.kafka.config.KafkaListenerConfigUtils;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.EmbeddedKafkaKraftBroker;
+import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
@@ -56,6 +55,7 @@
 
 @SpringJUnitConfig
 @DirtiesContext
+@EmbeddedKafka(partitions = 1, topics = "default-listener.tests")
 public class KafkaListenerTests {
 
 	private static final String TEST_TOPIC = "default-listener.tests";
@@ -118,15 +118,8 @@ public void testImplicitKafkaHandlerAnnotation() throws Exception {
 	@EnableKafka
 	public static class Config {
 
-		@Bean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
-		public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry() {
-			return new KafkaListenerEndpointRegistry();
-		}
-
-		@Bean
-		public EmbeddedKafkaBroker embeddedKafka() {
-			return new EmbeddedKafkaKraftBroker(1, 1, TEST_TOPIC);
-		}
+		@Autowired
+		EmbeddedKafkaBroker broker;
 
 		@Bean
 		public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
@@ -143,8 +136,10 @@ public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
 
 		@Bean
 		public Map<String, Object> consumerConfigs() {
-			Map<String, Object> consumerProps =
-					KafkaTestUtils.consumerProps("myDefaultListenerGroup", "false", embeddedKafka());
+			Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
+					this.broker.getBrokersAsString(),
+					"myDefaultListenerGroup",
+					"false");
 			return consumerProps;
 		}
 
@@ -160,7 +155,7 @@ public ProducerFactory<Integer, String> producerFactory() {
 
 		@Bean
 		public Map<String, Object> producerConfigs() {
-			return KafkaTestUtils.producerProps(embeddedKafka());
+			return KafkaTestUtils.producerProps(this.broker.getBrokersAsString());
 		}
 
 		@Component