Skip to content

GH-3807: Necessity of KafkaHandler on single method class #3865

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 6 commits into from
Apr 29, 2025
Merged

GH-3807: Necessity of KafkaHandler on single method class #3865

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9511818
spring-projectsGH-3807: Necessity of KafkaHandler on single method class
chickenchickenlove Apr 24, 2025
13ee0d1
Addressing PR review
chickenchickenlove Apr 25, 2025
19f9371
Addressing PR review
chickenchickenlove Apr 25, 2025
62f82f7
Revert to a52faccd
chickenchickenlove Apr 27, 2025
c322d68
Addressing PR review
chickenchickenlove Apr 27, 2025
f395cce
Addressing PR review
chickenchickenlove Apr 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 18 additions & 2 deletions ...n/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -411,8 +412,23 @@ public Object postProcessAfterInitialization(final Object bean, final String bea
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);
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.
// See, GH-3807 for more details.
Method method = publicMethods.iterator().next();
for (KafkaListener classLevelListener : classLevelListeners) {
processKafkaListener(classLevelListener, method, bean, beanName);
}
}
else {
List<Method> multiMethods = new ArrayList<>(methodsWithHandler);
processMultiMethodListeners(classLevelListeners, multiMethods, targetClass, bean, beanName);
}
}
}
}
Expand Down
20 changes: 20 additions & 0 deletions spring-kafka/src/test/java/org/springframework/kafka/annotation/AliasPropertiesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AliasFor;
import org.springframework.kafka.annotation.AliasPropertiesTests.Config.ClassAndMethodLevelListener;
import org.springframework.kafka.annotation.AliasPropertiesTests.Config.ClassLevelListenerWithPublicMethodWithoutHandlerAnnotation;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerConfigUtils;
Expand All @@ -58,6 +59,7 @@
* @author Gary Russell
* @author Artem Bilan
* @author Soby Chacko
* @author Sanghyeok An
*
* @since 2.2
*
Expand All @@ -81,12 +83,16 @@ public class AliasPropertiesTests {
@Autowired
private ClassAndMethodLevelListener repeatable;

@Autowired
private ClassLevelListenerWithPublicMethodWithoutHandlerAnnotation withPublicMethod;

@Test
public void testAliasFor() throws Exception {
this.template.send("alias.tests", "foo");
assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.classLevel.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.repeatable.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.withPublicMethod.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.config.kafkaListenerEndpointRegistry()).isSameAs(this.kafkaListenerEndpointRegistry);
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("onMethodInConfigClass").getGroupId())
.isEqualTo("onMethodInConfigClass.Config.listen1");
Expand All @@ -102,6 +108,8 @@ public void testAliasFor() throws Exception {
.isEqualTo("onMethodRepeatable2.RepeatableClassAndMethodLevelListener.listen1");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("onClassRepeatable2").getGroupId())
.isEqualTo("onClassRepeatable2.RepeatableClassAndMethodLevelListener");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("onSinglePublicMethod").getGroupId())
.isEqualTo("onSinglePublicMethod.ClassLevelListenerWithPublicMethodWithoutHandlerAnnotation");
assertThat(Config.orderedCalledFirst.get()).isTrue();
}

Expand Down Expand Up @@ -220,6 +228,18 @@ public void listen1(String in) {

}

@Component
@MyListener(id = "onSinglePublicMethod", value = "alias.tests")
public static class ClassLevelListenerWithPublicMethodWithoutHandlerAnnotation {

private final CountDownLatch latch = new CountDownLatch(1);

public void listen(String in) {
this.latch.countDown();
}

}

public static class OrderedEnhancer implements AnnotationEnhancer, Ordered {

private final AtomicBoolean orderedCalledFirst;
Expand Down
234 changes: 234 additions & 0 deletions spring-kafka/src/test/java/org/springframework/kafka/annotation/KafkaListenerTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,234 @@
/*
* Copyright 2016-2025 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.
* You may obtain a copy of the License at
*
* 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,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.springframework.kafka.annotation;

import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.jupiter.api.Test;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListenerTests.Config.ClassAndMethodLevelListener;
import org.springframework.kafka.annotation.KafkaListenerTests.Config.ClassLevelListenerWithDoublePublicMethod;
import org.springframework.kafka.annotation.KafkaListenerTests.Config.ClassLevelListenerWithSinglePublicMethod;
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.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.context.EmbeddedKafka;
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;

import static org.assertj.core.api.Assertions.assertThat;

/**
* @author Sanghyeok An
*
* @since 4.0
*
*/

@SpringJUnitConfig
@DirtiesContext
@EmbeddedKafka(partitions = 1, topics = "default-listener.tests")
public class KafkaListenerTests {

private static final String TEST_TOPIC = "default-listener.tests";

@Autowired
private KafkaTemplate<Integer, String> template;

@Autowired
private Config config;

@Autowired
private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

@Autowired
private ClassAndMethodLevelListener classLevel;

@Autowired
private ClassLevelListenerWithSinglePublicMethod classLevelListenerWithSinglePublicMethod;

@Autowired
private OtherClassLevelListenerWithSinglePublicMethod otherClassLevelListenerWithSinglePublicMethod;

@Autowired
private ClassLevelListenerWithDoublePublicMethod classLevelListenerWithDoublePublicMethod;

@Autowired
private ClassLevelListenerWithSinglePublicMethodAndPrivateMethod classLevelListenerWithSinglePublicMethodAndPrivateMethod;

@Test
public void testImplicitKafkaHandlerAnnotation() throws Exception {
// GIVEN
// See, bean configuration.

// WHEN
this.template.send(TEST_TOPIC, "foo");

// THEN
assertThat(this.classLevel.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.classLevelListenerWithSinglePublicMethod.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.otherClassLevelListenerWithSinglePublicMethod.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.classLevelListenerWithSinglePublicMethodAndPrivateMethod.latch.await(10, TimeUnit.SECONDS)).isTrue();

assertThat(this.classLevelListenerWithDoublePublicMethod.latch.await(10, TimeUnit.SECONDS)).isFalse();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not OK.
I'm really against any blocking in the tests for nothing.
I mean this always going to add up 10 seconds to the whole test suite execution.
Now imaging we have like 3000 tests in the project and every one would contribute some delay.

I believe the configuration when we have ambiguity with several public methods has to be reject with an IllegalStateException.
No silent ignoring to make am impression that my configuration is OK.

Copy link
Contributor Author

@chickenchickenlove chickenchickenlove Apr 25, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@artembilan thanks for your comments. 🙇‍♂️

This is not OK.
I'm really against any blocking in the tests for nothing.
I mean this always going to add up 10 seconds to the whole test suite execution.
Now imaging we have like 3000 tests in the project and every one would contribute some delay.

I agree.
I will remove this case from test codes.

I believe the configuration when we have ambiguity with several public methods has to be reject with an IllegalStateException.

I understand your point clearly.
However, The important thing is that the changes I made in this PR are not related to the case where there are multiple public methods.
So we don't need to consider ambiguous conditions because in that case any public method will not be registered as kafka listeners.

if (hasClassLevelListeners) {
  if (methodsWithHandler.isEmpty() && publicMethods.size() == 1 && !hasMethodLevelListeners) {
    // added this time.
  }
  else {
    // original	
  }
}

I think we would be better not to throw an error if there are several public methods when there is no @KafkaHandler method.
This is because after registering the class as a bean using @KafkaListener, there might be cases where other beans inject it and call its public methods. For example, getter methods.
If there are already users using it that way, it would be difficult to ensure backward compatibility.

@KafkaListener
class MyTakser {
  
  // This method is for @KafkaHandler implicitly.
  public void listen (String message) {
    // do something with kafka messages.
  }

  // This method is not for @KafkaHandler.
  public void getCurrentState() {
    // return current state
  }
  
  ...
}

IMHO, If we should something to do for this, I think it would be better to notify by using warning log.
For example,

The bean {BEAN_NAME} which has @KafkaListener is registered without a proper listener method.

WDYT?

Copy link
Member

@artembilan artembilan Apr 25, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think I see the picture better now. What if our class is extended and new methods appear there? Which one to choose from hierarchy then?

I am sorry to say that , but I feel more like better to leave it as is and rely only on the @KafkaHandler presence .

The issue simply could be turned into docs explaining why we require this or other annotation. We really cannot predict what purpose of the method is, so being explicit is much better, than trying to come up with other restrictions. I have totally ignored the fact that class could be extended 🤦.

And between us: I’m not a fun of annotations in Java. 😉

Still, that is my opinion on the problem of this issue, so I’m opened for other arguments to continue your effort with the code.

CC @sobychacko

Copy link
Contributor Author

@chickenchickenlove chickenchickenlove Apr 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@artembilan
Thank you for your detailed explanation 🙇‍♂️
I prefer to use explicit one as well. annotations too, because of implicit 😉.

No worries at all — I'm happy to revert all the commits I made if needed! 😄

How about reconsidering it from this perspective?

1. In the case of a class with only a single public method

I believe we don't need to consider class extension in this case.
If the class was originally designed with a single public method intended to act as an implicit @KafkaHandler, it should be fine to assume that even if the class is extended, the same usage will be maintained.

If an extended class introduces a new public method, it will be ignored because there are two public methods.
If an extended class override parent's public method for a completely different purpose, then it should be considered the user's responsibility.

2. In the case of a class with multiple public methods

If there are multiple public methods and no explicit @KafkaHandler annotation is present, no method would be registered as a listener. it is current registering logic.

Therefore, I believe we don't need to worry about this case either.
For these reasons, I think this PR still can have value.

After you read my comments and still believe this PR is invalid, I'm happy to revert all the commits and will create commits to update class-level-kafkalistener.adoc!
Please, feel free to talk your opinion.

Thanks for your time 🙇‍♂️

assertThat(this.classLevelListenerWithDoublePublicMethod.latch.getCount()).isEqualTo(ClassLevelListenerWithDoublePublicMethod.INIT_LATCH_COUNT);

assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("classAndMethodLevelListener").getGroupId())
.isEqualTo("classAndMethodLevelListener");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("classLevelListenerWithSinglePublicMethod").getGroupId())
.isEqualTo("classLevelListenerWithSinglePublicMethod");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("otherClassLevelListenerWithSinglePublicMethod").getGroupId())
.isEqualTo("otherClassLevelListenerWithSinglePublicMethod");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("classLevelListenerWithDoublePublicMethod").getGroupId())
.isEqualTo("classLevelListenerWithDoublePublicMethod");
assertThat(this.kafkaListenerEndpointRegistry.getListenerContainer("classLevelListenerWithSinglePublicMethodAndPrivateMethod").getGroupId())
.isEqualTo("classLevelListenerWithSinglePublicMethodAndPrivateMethod");

}

@Configuration
@EnableKafka
public static class Config {

@Autowired
EmbeddedKafkaBroker broker;

@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}

@Bean
public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"myDefaultListenerGroup",
"false");
return consumerProps;
}

@Bean
public KafkaTemplate<Integer, String> template() {
return new KafkaTemplate<>(producerFactory());
}

@Bean
public ProducerFactory<Integer, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}

@Bean
public Map<String, Object> producerConfigs() {
return KafkaTestUtils.producerProps(this.broker.getBrokersAsString());
}

@Component
@KafkaListener(id = "classAndMethodLevelListener", topics = "default-listener.tests")
public static class ClassAndMethodLevelListener {

final CountDownLatch latch = new CountDownLatch(1);

@KafkaHandler
public void listen(String in) {
this.latch.countDown();
}
}

@Component
@KafkaListener(id = "classLevelListenerWithSinglePublicMethod", topics = TEST_TOPIC)
public static class ClassLevelListenerWithSinglePublicMethod {

final CountDownLatch latch = new CountDownLatch(1);

public void listen(String in) {
this.latch.countDown();
}
}

@Component
@KafkaListener(id = "otherClassLevelListenerWithSinglePublicMethod", topics = TEST_TOPIC)
public static class OtherClassLevelListenerWithSinglePublicMethod {

final CountDownLatch latch = new CountDownLatch(1);

public void listen(String in) {
this.latch.countDown();
}
}

@Component
@KafkaListener(id = "classLevelListenerWithDoublePublicMethod", topics = TEST_TOPIC)
public static class ClassLevelListenerWithDoublePublicMethod {

public final static int INIT_LATCH_COUNT = 2;

final CountDownLatch latch = new CountDownLatch(INIT_LATCH_COUNT);

public void listen1(String in) {
this.latch.countDown();
}

public void listen2(String in) {
this.latch.countDown();
}
}

@Component
@KafkaListener(id = "classLevelListenerWithSinglePublicMethodAndPrivateMethod", topics = TEST_TOPIC)
public static class ClassLevelListenerWithSinglePublicMethodAndPrivateMethod {

final CountDownLatch latch = new CountDownLatch(1);

public void listen1(String in) {
this.latch.countDown();
}

private void listen2(String in) {
this.latch.countDown();
}

private void listen3(String in) {
this.latch.countDown();
}

}

}

}