SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2941: Kafka Streams Component beans composition

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2941

* Introduce function composition for Kafka Streams Component beans
This commit is contained in:
Soby Chacko
2024-04-30 20:55:16 -04:00
parent f230d873c7
commit 1d241583c3
3 changed files with 82 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java
View File

@@ -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<String, ResolvableType> resolvableTypes = buildTypeMap(resolvableType,
Map<String, ResolvableType> 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<String> outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs());
final Iterator<String> outboundDefinitionIterator = outputs.iterator();

24
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
View File

@@ -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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> 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<KafkaStreamsBindableProxyFactory> proxyFactory =
@@ -88,4 +97,5 @@ public class KafkaStreamsFunctionProcessorInvoker {
}
}
}
}

53
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java
View File

@@ -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<String, String> consumer1;
private static Consumer<String, String> consumer2;
@@ -70,6 +70,7 @@ class KafkaStreamsComponentBeansTests {
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, String> 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<KStream<String, String>> {
@Override
public void accept(KStream<String, String> integerStringKStream) {
integerStringKStream.foreach((s1, s2) -> LATCH_1.countDown());
}
}
@Component("bazz")
@EnableAutoConfiguration
public static class BiFunctionAsComponent implements BiFunction<KStream<String, String>, KStream<String, String>, KStream<String, String>> {