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

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,

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<String> kafkaStreamsMethodNames = new HashSet<>(kafkaStreamsOnlyResolvableTypes.keySet());
 		kafkaStreamsMethodNames.addAll(this.resolvableTypeMap.keySet());

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) -> {

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");
+	}
+}