From 2d2d4f2f5e8383288cb735f2aa2995f9dbed5add Mon Sep 17 00:00:00 2001 From: jamesforward Date: Mon, 22 May 2023 20:15:23 +0100 Subject: [PATCH] Multiple functions in Kafka Streams binder * Trim functionUnits so we can declare them in a multi-line fashion, as currently whitespace is still considered. * Polishing Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2742 --- .../streams/KafkaStreamsBinderUtils.java | 9 ++++ ...KafkaStreamsFunctionBeanPostProcessor.java | 5 ++- .../KafkaStreamsFunctionProcessorInvoker.java | 6 ++- .../streams/KafkaStreamsBinderUtilsTests.java | 43 +++++++++++++++++++ 4 files changed, 59 insertions(+), 4 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtilsTests.java diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index 29ef31c98..cde79feaa 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -66,6 +66,7 @@ import org.springframework.util.StringUtils; * * @author Soby Chacko * @author Gary Russell + * @author James Forward */ public final class KafkaStreamsBinderUtils { @@ -86,6 +87,14 @@ public final class KafkaStreamsBinderUtils { return Arrays.stream(methods).filter(m -> m.getName().equals(key)).findFirst(); } + public static String[] deriveFunctionUnits(String definition) { + if (!StringUtils.hasText(definition)) { + return new String[]{}; + } + final String[] rawSplitDefinition = definition.split(";"); + return Arrays.stream(rawSplitDefinition).map(String::trim).toArray(String[]::new); + } + static void prepareConsumerBinding(String name, String group, ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 655b06168..8276eaecd 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -52,10 +52,11 @@ import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderU import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; import org.springframework.util.ClassUtils; -import org.springframework.util.StringUtils; /** * @author Soby Chacko + * @author James Forward + * * @since 2.2.0 */ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, BeanFactoryAware { @@ -110,7 +111,7 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; final String definition = streamFunctionProperties.getDefinition(); - final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{}; + final String[] functionUnits = KafkaStreamsBinderUtils.deriveFunctionUnits(definition); final Set kafkaStreamsMethodNames = new HashSet<>(kafkaStreamsOnlyResolvableTypes.keySet()); kafkaStreamsMethodNames.addAll(this.resolvableTypeMap.keySet()); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java index 9669686a8..ba9333702 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java @@ -23,13 +23,15 @@ import java.util.Optional; import jakarta.annotation.PostConstruct; +import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsBinderUtils; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsFunctionProcessor; import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; -import org.springframework.util.StringUtils; /** * @author Soby Chacko + * @author James Forward + * * @since 2.1.0 */ public class KafkaStreamsFunctionProcessorInvoker { @@ -54,7 +56,7 @@ public class KafkaStreamsFunctionProcessorInvoker { @PostConstruct void invoke() { final String definition = streamFunctionProperties.getDefinition(); - final String[] functionUnits = StringUtils.hasText(definition) ? definition.split(";") : new String[]{}; + final String[] functionUnits = KafkaStreamsBinderUtils.deriveFunctionUnits(definition); if (functionUnits.length == 0) { resolvableTypeMap.forEach((key, value) -> { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtilsTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtilsTests.java new file mode 100644 index 000000000..60b10f2c6 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtilsTests.java @@ -0,0 +1,43 @@ +/* + * Copyright 2023-2023 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.streams; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for {@link KafkaStreamsBinderUtils} functions + * + * @author James Forward + * @author Soby Chacko + */ +class KafkaStreamsBinderUtilsTests { + + @Test + void testDeriveFunctionUnitsWithVariedSpacings() { + final String definition = "firstFunction; secondFunction; thirdFunction;fourthFunction"; + + final String[] functionUnits = KafkaStreamsBinderUtils.deriveFunctionUnits(definition); + + assertThat(functionUnits.length).isEqualTo(4); + assertThat(functionUnits[0]).isEqualTo("firstFunction"); + assertThat(functionUnits[1]).isEqualTo("secondFunction"); + assertThat(functionUnits[2]).isEqualTo("thirdFunction"); + assertThat(functionUnits[3]).isEqualTo("fourthFunction"); + } +}