diff --git a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
index c52e1eff..2fddce27 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java
@@ -362,7 +362,7 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 								AnnotationUtils.findAnnotation(method, KafkaHandler.class) != null);
 				multiMethods.addAll(methodsWithHandler);
 			}
-			if (annotatedMethods.isEmpty()) {
+			if (annotatedMethods.isEmpty() && !hasClassLevelListeners) {
 				this.nonAnnotatedClasses.add(bean.getClass());
 				this.logger.trace(() -> "No @KafkaListener annotations found on bean type: " + bean.getClass());
 			}
diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
index 69fa2a7a..0a62c404 100644
--- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
+++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
@@ -185,7 +185,7 @@ import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
 		"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
 		"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
 		"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41", "annotated42",
-		"annotated43", "annotated43reply"})
+		"annotated43", "annotated43reply" })
 @TestPropertySource(properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10")
 public class EnableKafkaIntegrationTests {
 
@@ -1009,6 +1009,12 @@ public class EnableKafkaIntegrationTests {
 		this.registry.setAlwaysStartAfterRefresh(true);
 	}
 
+	@Test
+	void classLevelTwoInstancesSameClass() {
+		assertThat(this.registry.getListenerContainer("multiTwoOne")).isNotNull();
+		assertThat(this.registry.getListenerContainer("multiTwoTwo")).isNotNull();
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)
@@ -1731,6 +1737,16 @@ public class EnableKafkaIntegrationTests {
 			return new ProtoListener();
 		}
 
+		@Bean
+		MultiListenerTwoInstances multiInstanceOne() {
+			return new MultiListenerTwoInstances("multiTwoOne");
+		}
+
+		@Bean
+		MultiListenerTwoInstances multiInstanceTwo() {
+			return new MultiListenerTwoInstances("multiTwoTwo");
+		}
+
 	}
 
 	static class ProtoListener {
@@ -2461,6 +2477,25 @@ public class EnableKafkaIntegrationTests {
 
 	}
 
+	@KafkaListener(id = "#{__listener.id}", topics = "multiWithTwoInstances", autoStartup = "false")
+	static class MultiListenerTwoInstances {
+
+		private final String id;
+
+		MultiListenerTwoInstances(String id) {
+			this.id = id;
+		}
+
+		public String getId() {
+			return this.id;
+		}
+
+		@KafkaHandler
+		void listen(String in) {
+		}
+
+	}
+
 	public interface Bar {
 
 		String getBar();