GH-4170 : Add KafkaListener Validation (Allow @Topic or @TopicPartition) (#4172)

moonyoungCHAE · web-flow · commit a0b5b70cfc06 · 2025-12-03T16:53:14.000-05:00
Fixes #4170 **Auto-cherry-pick to `3.3.x`** * add topic validation of kafka listener * refactor: allow topic pattern for validtion Signed-off-by: moonyoungCHAE <xpf_fl@naver.com>
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
@@ -663,9 +663,12 @@ private void processKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?> en
 		endpoint.setBean(bean);
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));
+
+		assertTopic(kafkaListener);
 		endpoint.setTopicPartitions(tps);
 		endpoint.setTopics(topics);
 		endpoint.setTopicPattern(resolvePattern(kafkaListener));
+
 		endpoint.setClientIdPrefix(resolveExpressionAsString(kafkaListener.clientIdPrefix(), "clientIdPrefix"));
 		endpoint.setListenerInfo(resolveExpressionAsBytes(kafkaListener.info(), "info"));
 		String group = kafkaListener.containerGroup();
@@ -856,6 +859,20 @@ private String getEndpointGroupId(KafkaListener kafkaListener, @Nullable String
 		return groupId;
 	}
 
+	private void assertTopic(KafkaListener kafkaListener) {
+		int count = 0;
+		if (!kafkaListener.topicPattern().isEmpty()) {
+			count++;
+		}
+		if (kafkaListener.topics().length > 0) {
+			count++;
+		}
+		if (kafkaListener.topicPartitions().length > 0) {
+			count++;
+		}
+		Assert.state(count == 1, "Only one of @Topic or @TopicPartition or @TopicPattern must be provided");
+	}
+
 	private TopicPartitionOffset[] resolveTopicPartitions(KafkaListener kafkaListener) {
 		TopicPartition[] topicPartitions = kafkaListener.topicPartitions();
 		List<TopicPartitionOffset> result = new ArrayList<>();
