refactor: allow topic pattern for validtion

Signed-off-by: moonyoungCHAE <[email protected]>

Author: moonyoungCHAE

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

@@ -666,10 +666,11 @@ private void processKafkaListenerAnnotation(MethodKafkaListenerEndpoint<?, ?> en
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setGroupId(getEndpointGroupId(kafkaListener, endpoint.getId()));
 
-		Assert.state((topics.length > 0) ^ (tps.length > 0), "Only one of @Topic or @TopicPartition is allowed");
+		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();
@@ -860,6 +861,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<>();