Skip to content

Commit 2966c7d

Browse files
garyrussellartembilan
authored andcommitted
GH-2451: Fix Class Level Listener Multi Instances
Resolves #2451 Classes with class level `@KafkaListener` were incorrectly added to the `nonAnnotatedClasses` set, preventing multiple instances of the same class to be registered as listeners. **cherry-pick to 2.9.x, 2.8.x** * Fix CheckStyle.
1 parent 9e07035 commit 2966c7d

File tree

2 files changed

+37
-2
lines changed

2 files changed

+37
-2
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -377,7 +377,7 @@ public Object postProcessAfterInitialization(final Object bean, final String bea
377377
AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
378378
multiMethods.addAll(methodsWithHandler);
379379
}
380-
if (annotatedMethods.isEmpty()) {
380+
if (annotatedMethods.isEmpty() && !hasClassLevelListeners) {
381381
this.nonAnnotatedClasses.add(bean.getClass());
382382
this.logger.trace(() -> "No @KafkaListener annotations found on bean type: " + bean.getClass());
383383
}

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

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -188,7 +188,7 @@
188188
"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
189189
"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
190190
"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41", "annotated42",
191-
"annotated43", "annotated43reply"})
191+
"annotated43", "annotated43reply" })
192192
@TestPropertySource(properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10")
193193
public class EnableKafkaIntegrationTests {
194194

@@ -1012,6 +1012,12 @@ void proto(@Autowired ApplicationContext context) {
10121012
this.registry.setAlwaysStartAfterRefresh(true);
10131013
}
10141014

1015+
@Test
1016+
void classLevelTwoInstancesSameClass() {
1017+
assertThat(this.registry.getListenerContainer("multiTwoOne")).isNotNull();
1018+
assertThat(this.registry.getListenerContainer("multiTwoTwo")).isNotNull();
1019+
}
1020+
10151021
@Configuration
10161022
@EnableKafka
10171023
@EnableTransactionManagement(proxyTargetClass = true)
@@ -1742,6 +1748,16 @@ ProtoListener proto() {
17421748
return new ProtoListener();
17431749
}
17441750

1751+
@Bean
1752+
MultiListenerTwoInstances multiInstanceOne() {
1753+
return new MultiListenerTwoInstances("multiTwoOne");
1754+
}
1755+
1756+
@Bean
1757+
MultiListenerTwoInstances multiInstanceTwo() {
1758+
return new MultiListenerTwoInstances("multiTwoTwo");
1759+
}
1760+
17451761
}
17461762

17471763
static class ProtoListener {
@@ -2483,6 +2499,25 @@ public String bar(@Payload(required = false) KafkaNull nul,
24832499

24842500
}
24852501

2502+
@KafkaListener(id = "#{__listener.id}", topics = "multiWithTwoInstances", autoStartup = "false")
2503+
static class MultiListenerTwoInstances {
2504+
2505+
private final String id;
2506+
2507+
MultiListenerTwoInstances(String id) {
2508+
this.id = id;
2509+
}
2510+
2511+
public String getId() {
2512+
return this.id;
2513+
}
2514+
2515+
@KafkaHandler
2516+
void listen(String in) {
2517+
}
2518+
2519+
}
2520+
24862521
public interface Bar {
24872522

24882523
String getBar();

0 commit comments

Comments
 (0)