From 82a3306cb9710832221403c431947a2184a47b11 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 1 Oct 2021 11:07:14 -0400 Subject: [PATCH] GH-1157: Issues with Kafka Streams and Kotlin Kafka Streams binder erroneously tries to parse regular non Kafka streams Kotlin function registrations. Ignore function beans ending in _registration in Kafka Streams binder. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1157 --- .../streams/function/KafkaStreamsFunctionBeanPostProcessor.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 0e7e99785..153f19d42 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -29,6 +29,7 @@ import java.util.function.BiConsumer; import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; +import java.util.regex.Pattern; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -96,6 +97,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, Stream.concat(Stream.of(biFunctionNames), Stream.of(biConsumerNames))); final List collect = concat.collect(Collectors.toList()); collect.removeIf(s -> Arrays.stream(EXCLUDE_FUNCTIONS).anyMatch(t -> t.equals(s))); + collect.removeIf(Pattern.compile(".*_registration").asPredicate()); onlySingleFunction = collect.size() == 1; collect.stream()