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

Fixes #4170

* add topic validation of kafka listener

* refactor: allow topic pattern for validtion

Signed-off-by: moonyoungCHAE <xpf_fl@naver.com>
(cherry picked from commit a0b5b70)

Author: moonyoungCHAE

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

@@ -666,9 +666,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();
@@ -859,6 +862,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<>();