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

GH-2843: Kafka Streams binder component bean issue

Browse Source 
Fix automatic Serde detection for branched function components

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2843
This commit is contained in:
Georg Friedrich
2023-11-05 22:15:32 +01:00
committed by Soby Chacko
parent d130b46110
commit ba0924acc2
2 changed files with 73 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -68,6 +68,7 @@ import org.springframework.util.CollectionUtils;
/**
* @author Soby Chacko
* @author Byungjun You
* @author Georg Friedrich
* @since 2.2.0
*/
public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderProcessor implements BeanFactoryAware {
@@ -122,8 +123,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
final Iterator<String> iterator = inputs.iterator();
populateResolvableTypeMap(firstMethodParameter, resolvableTypeMap, iterator, method, functionName);
final Class<?> outputRawclass = currentOutputGeneric.getRawClass();
traverseReturnTypeForComponentBeans(resolvableTypeMap, currentOutputGeneric, inputs, iterator, outputRawclass);
traverseReturnTypeForComponentBeans(resolvableTypeMap, currentOutputGeneric, inputs, iterator);
}
else if (resolvableType != null && resolvableType.getRawClass() != null) {
int inputCount = 1;
@@ -171,8 +171,9 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
}
private void traverseReturnTypeForComponentBeans(Map<String, ResolvableType> resolvableTypeMap, ResolvableType currentOutputGeneric,
Set<String> inputs, Iterator<String> iterator, Class<?> outputRawclass) {
if (outputRawclass != null && !outputRawclass.equals(Void.TYPE)) {
Set<String> inputs, Iterator<String> iterator) {
final Class<?> outputRawclass = currentOutputGeneric.getRawClass();
if (outputRawclass != null && !outputRawclass.equals(Void.TYPE) || currentOutputGeneric.isArray()) {
ResolvableType iterableResType = currentOutputGeneric;
int i = 1;
// Traverse through the return signature.
@@ -184,7 +185,9 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro
iterableResType = iterableResType.getGeneric(1);
i++;
}
if (iterableResType.getRawClass() != null && KStream.class.isAssignableFrom(iterableResType.getRawClass())) {
if (iterableResType.getRawClass() != null && KStream.class.isAssignableFrom(iterableResType.getRawClass())
|| iterableResType.isArray() && iterableResType.getComponentType().getRawClass() != null
&& KStream.class.isAssignableFrom(iterableResType.getComponentType().getRawClass())) {
resolvableTypeMap.put(OUTBOUND, iterableResType);
}
}

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

@@ -52,8 +52,9 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Soby Chacko
* @author Georg Friedrich
*/
@EmbeddedKafka(topics = {"testFunctionComponent-out", "testBiFunctionComponent-out", "testCurriedFunctionWithFunctionTerminal-out"})
@EmbeddedKafka(topics = {"testFunctionComponent-out-0", "testFunctionComponent-out-1", "testBiFunctionComponent-out", "testCurriedFunctionWithFunctionTerminal-out"})
class KafkaStreamsComponentBeansTests {
private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();
@@ -61,6 +62,7 @@ class KafkaStreamsComponentBeansTests {
private static Consumer<String, String> consumer1;
private static Consumer<String, String> consumer2;
private static Consumer<String, String> consumer3;
private static Consumer<String, String> consumer4;
private final static CountDownLatch LATCH_1 = new CountDownLatch(1);
private final static CountDownLatch LATCH_2 = new CountDownLatch(2);
@@ -74,7 +76,7 @@ class KafkaStreamsComponentBeansTests {
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer1 = cf.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer1, "testFunctionComponent-out");
embeddedKafka.consumeFromEmbeddedTopics(consumer1, "testFunctionComponent-out-0");
Map<String, Object> consumerProps1 = KafkaTestUtils.consumerProps("group-x", "false",
embeddedKafka);
@@ -82,7 +84,7 @@ class KafkaStreamsComponentBeansTests {
consumerProps1.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
DefaultKafkaConsumerFactory<String, String> cf1 = new DefaultKafkaConsumerFactory<>(consumerProps1);
consumer2 = cf1.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer2, "testBiFunctionComponent-out");
embeddedKafka.consumeFromEmbeddedTopics(consumer2, "testFunctionComponent-out-1");
Map<String, Object> consumerProps2 = KafkaTestUtils.consumerProps("group-y", "false",
embeddedKafka);
@@ -90,7 +92,15 @@ class KafkaStreamsComponentBeansTests {
consumerProps2.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
DefaultKafkaConsumerFactory<String, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps2);
consumer3 = cf2.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer3, "testCurriedFunctionWithFunctionTerminal-out");
embeddedKafka.consumeFromEmbeddedTopics(consumer3, "testBiFunctionComponent-out");
Map<String, Object> consumerProps3 = KafkaTestUtils.consumerProps("group-z", "false",
embeddedKafka);
consumerProps3.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps3.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
DefaultKafkaConsumerFactory<String, String> cf3 = new DefaultKafkaConsumerFactory<>(consumerProps3);
consumer4 = cf3.createConsumer();
embeddedKafka.consumeFromEmbeddedTopics(consumer4, "testCurriedFunctionWithFunctionTerminal-out");
}
@AfterAll
@@ -98,6 +108,7 @@ class KafkaStreamsComponentBeansTests {
consumer1.close();
consumer2.close();
consumer3.close();
consumer4.close();
}
@Test
@@ -108,7 +119,7 @@ class KafkaStreamsComponentBeansTests {
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.foo-in-0.destination=testFunctionComponent-in",
"--spring.cloud.stream.bindings.foo-out-0.destination=testFunctionComponent-out",
"--spring.cloud.stream.bindings.foo-out-0.destination=testFunctionComponent-out-0",
"--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);
@@ -117,7 +128,7 @@ class KafkaStreamsComponentBeansTests {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("testFunctionComponent-in");
template.sendDefault("foobar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer1, "testFunctionComponent-out");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer1, "testFunctionComponent-out-0");
assertThat(cr.value().contains("foobarfoobar")).isTrue();
}
finally {
@@ -126,6 +137,36 @@ class KafkaStreamsComponentBeansTests {
}
}
@Test
void functionComponentWithBranching() {
SpringApplication app = new SpringApplication(FunctionAsComponentWithBranching.class);
app.setWebApplicationType(WebApplicationType.NONE);
try (ConfigurableApplicationContext ignored = app.run(
"--server.port=0",
"--spring.jmx.enabled=false",
"--spring.cloud.stream.bindings.branchedFoo-in-0.destination=testFunctionBranchingComponent-in",
"--spring.cloud.stream.bindings.branchedFoo-out-0.destination=testFunctionComponent-out-0",
"--spring.cloud.stream.bindings.branchedFoo-out-1.destination=testFunctionComponent-out-1",
"--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("testFunctionBranchingComponent-in");
template.sendDefault(0, "foo");
template.sendDefault(1, "bar");
ConsumerRecord<String, String> cr = KafkaTestUtils.getSingleRecord(consumer1, "testFunctionComponent-out-0");
assertThat(cr.value().contains("foo")).isTrue();
ConsumerRecord<String, String> cr2 = KafkaTestUtils.getSingleRecord(consumer2, "testFunctionComponent-out-1");
assertThat(cr2.value().contains("bar")).isTrue();
}
finally {
pf.destroy();
}
}
}
@Test
void consumerComponent() throws Exception {
SpringApplication app = new SpringApplication(ConsumerAsComponent.class);
@@ -170,7 +211,7 @@ class KafkaStreamsComponentBeansTests {
template.sendDefault("foobar");
template.setDefaultTopic("testBiFunctionComponent-in-1");
template.sendDefault("foobar");
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer2, Duration.ofSeconds(10), 2);
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer3, Duration.ofSeconds(10), 2);
assertThat(records.count()).isEqualTo(2);
records.forEach(stringStringConsumerRecord -> assertThat(stringStringConsumerRecord.value().contains("foobar")).isTrue());
}
@@ -260,7 +301,7 @@ class KafkaStreamsComponentBeansTests {
template.sendDefault("foobar");
template.setDefaultTopic("testCurriedFunctionWithFunctionTerminal-in-2");
template.sendDefault("foobar");
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer3, Duration.ofSeconds(10), 3);
final ConsumerRecords<String, String> records = KafkaTestUtils.getRecords(consumer4, Duration.ofSeconds(10), 3);
assertThat(records.count()).isEqualTo(3);
records.forEach(stringStringConsumerRecord -> assertThat(stringStringConsumerRecord.value().contains("foobar")).isTrue());
}
@@ -281,6 +322,22 @@ class KafkaStreamsComponentBeansTests {
}
}
@Component("branchedFoo")
@EnableAutoConfiguration
public static class FunctionAsComponentWithBranching implements Function<KStream<Integer, String>,
KStream<String, String>[]> {
@Override
public KStream<String, String>[] apply(KStream<Integer, String> stringIntegerKStream) {
return stringIntegerKStream.map((key, value) -> new KeyValue<>(key.toString(), value))
.split()
.branch((k, v) -> "1".equals(k))
.defaultBranch()
.values()
.toArray(new KStream[0]);
}
}
@Component("bar")
@EnableAutoConfiguration
public static class ConsumerAsComponent implements java.util.function.Consumer<KStream<Integer, String>> {