From 1d241583c34af6d33217d8793a8adfc7eba17944 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 30 Apr 2024 20:55:16 -0400 Subject: [PATCH] GH-2941: Kafka Streams Component beans composition Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2941 * Introduce function composition for Kafka Streams Component beans --- .../KafkaStreamsFunctionProcessor.java | 21 ++++++-- .../KafkaStreamsFunctionProcessorInvoker.java | 24 ++++++--- .../KafkaStreamsComponentBeansTests.java | 53 ++++++++++++++++--- 3 files changed, 82 insertions(+), 16 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index d510d29ca..e846781e2 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -254,8 +254,19 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory, Method method, ResolvableType outputResolvableType, String... composedFunctionNames) { - final Map resolvableTypes = buildTypeMap(resolvableType, + Map resolvableTypes; + + if (method != null && composedFunctionNames.length > 0) { // composed component methods + resolvableTypes = buildTypeMap(resolvableType, + kafkaStreamsBindableProxyFactory, method, composedFunctionNames[0]); + } + else if (method != null) { // non-composed component beans + resolvableTypes = buildTypeMap(resolvableType, kafkaStreamsBindableProxyFactory, method, functionName); + } + else { // all other cases + resolvableTypes = buildTypeMap(resolvableType, kafkaStreamsBindableProxyFactory, null, functionName); + } ResolvableType outboundResolvableType; if (outputResolvableType != null) { @@ -277,7 +288,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro biConsumer.accept(adaptedInboundArguments[0], adaptedInboundArguments[1]); } else if (method != null) { // Handling component functional beans - final Object bean = beanFactory.getBean(functionName); + final Object bean = composedFunctionNames.length > 0 ? beanFactory.getBean(composedFunctionNames[0]) + : beanFactory.getBean(functionName); if (Consumer.class.isAssignableFrom(bean.getClass())) { ((Consumer) bean).accept(adaptedInboundArguments[0]); } @@ -293,6 +305,9 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro result = ((Function) bean).apply(adaptedInboundArguments[0]); } result = handleCurriedFunctions(adaptedInboundArguments, result); + if (composedFunctionNames.length > 0) { + result = handleComposedFunctions(adaptedInboundArguments, result, composedFunctionNames); + } if (result != null) { final Set outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs()); final Iterator outboundDefinitionIterator = outputs.iterator(); 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 ba9333702..a8c018162 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 @@ -1,5 +1,5 @@ /* - * Copyright 2019-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -20,6 +20,7 @@ import java.lang.reflect.Method; import java.util.Arrays; import java.util.Map; import java.util.Optional; +import java.util.function.Consumer; import jakarta.annotation.PostConstruct; @@ -61,8 +62,10 @@ public class KafkaStreamsFunctionProcessorInvoker { if (functionUnits.length == 0) { resolvableTypeMap.forEach((key, value) -> { Optional proxyFactory = - Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(key)).findFirst(); - this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key, proxyFactory.get(), methods.get(key), null); + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(key)).findFirst(); + proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory -> + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key, + kafkaStreamsBindableProxyFactory, methods.get(key), null)); }); } @@ -74,10 +77,16 @@ public class KafkaStreamsFunctionProcessorInvoker { derivedNameFromComposed[0] = derivedNameFromComposed[0].concat(split); } Optional proxyFactory = - Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(derivedNameFromComposed[0])).findFirst(); - proxyFactory.ifPresent(kafkaStreamsBindableProxyFactory -> - this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]), - derivedNameFromComposed[0], kafkaStreamsBindableProxyFactory, methods.get(derivedNameFromComposed[0]), resolvableTypeMap.get(composedFunctions[composedFunctions.length - 1]), composedFunctions)); + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> + p.getFunctionName().equals(derivedNameFromComposed[0])).findFirst(); + + Method method = methods.isEmpty() ? null : methods.get(composedFunctions[0]); + Consumer kafkaStreamsBindableProxyFactoryConsumer = kafkaStreamsBindableProxyFactory -> + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(resolvableTypeMap.get(composedFunctions[0]), + derivedNameFromComposed[0], kafkaStreamsBindableProxyFactory, method, + resolvableTypeMap.get(composedFunctions[composedFunctions.length - 1]), composedFunctions); + + proxyFactory.ifPresent(kafkaStreamsBindableProxyFactoryConsumer); } else { Optional proxyFactory = @@ -88,4 +97,5 @@ public class KafkaStreamsFunctionProcessorInvoker { } } } + } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java index aa9595dcc..6be9def23 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -57,7 +57,7 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"testFunctionComponent-out-0", "testFunctionComponent-out-1", "testBiFunctionComponent-out", "testCurriedFunctionWithFunctionTerminal-out"}) class KafkaStreamsComponentBeansTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer1; private static Consumer consumer2; @@ -70,6 +70,7 @@ class KafkaStreamsComponentBeansTests { @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); @@ -105,10 +106,15 @@ class KafkaStreamsComponentBeansTests { @AfterAll public static void tearDown() { - consumer1.close(); - consumer2.close(); - consumer3.close(); - consumer4.close(); + closeConsumers(consumer1, consumer2, consumer3, consumer4); + } + + static void closeConsumers(Consumer... consumers) { + for (Consumer consumer : consumers) { + if (consumer != null) { + consumer.close(); + } + } } @Test @@ -137,6 +143,31 @@ class KafkaStreamsComponentBeansTests { } } + @Test + void functionComponentBeanComposedWithConsumer() throws InterruptedException { + SpringApplication app = new SpringApplication(FunctionAsComponent.class, StringConsumer.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.cloud.function.definition=foo|stringConsumer", + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.foostringConsumer-in-0.destination=fooStringConsumer-in", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("fooStringConsumer-in"); + template.sendDefault("foobar"); + Assert.isTrue(LATCH_1.await(10, TimeUnit.SECONDS), "foobar"); + } + finally { + pf.destroy(); + } + } + } + @Test void functionComponentWithBranching() { SpringApplication app = new SpringApplication(FunctionAsComponentWithBranching.class); @@ -349,6 +380,16 @@ class KafkaStreamsComponentBeansTests { } } + @Component("stringConsumer") + @EnableAutoConfiguration + public static class StringConsumer implements java.util.function.Consumer> { + + @Override + public void accept(KStream integerStringKStream) { + integerStringKStream.foreach((s1, s2) -> LATCH_1.countDown()); + } + } + @Component("bazz") @EnableAutoConfiguration public static class BiFunctionAsComponent implements BiFunction, KStream, KStream> {