From 309f588325628c23cd0cec36b858cea9c6c7dc80 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 3 Sep 2019 19:44:13 -0400 Subject: [PATCH] Addressing Kafka Streams multiple functions issues Fixing an issue that causes a race condition when multiple functions are present in a Kafka Streams application by isolating the responsible proxy factory per function and not shared. When multiple Kafka Streams functions are present in an application, it should be possible to set the application id per function. When an application provides a bean of type Serde, then the binder should try to introspect that bean to see if it can be matched for any inbound or outbound serialization. Adding tests to verify the changes. Adding docs. Resolves #734, #735, #736 --- docs/src/main/asciidoc/kafka-streams.adoc | 88 +++++++---- .../AbstractKafkaStreamsBinderProcessor.java | 27 +++- ...StreamsBinderSupportAutoConfiguration.java | 7 +- .../KafkaStreamsFunctionProcessor.java | 63 ++++---- ...StreamListenerSetupMethodOrchestrator.java | 2 +- .../kafka/streams/KeyValueSerdeResolver.java | 123 ++++++++++++--- .../KafkaStreamsBindableProxyFactory.java | 4 + ...KafkaStreamsFunctionAutoConfiguration.java | 7 +- .../KafkaStreamsFunctionProcessorInvoker.java | 14 +- ...aStreamsBinderConfigurationProperties.java | 13 ++ .../MultipleFunctionsInSameAppTests.java | 140 ++++++++++++++++++ .../function/SerdesProvidedAsBeansTests.java | 121 +++++++++++++++ 12 files changed, 519 insertions(+), 90 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java create mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index 701f029f7..de4a82990 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -494,9 +494,9 @@ If you only have one single processor in the application, then you can set this As a convenience, if you only have a single processor, you can also use `spring.application.name` as the property to delegate the application id. If you have multiple Kafka Streams processors in the application, then you need to set the application id per processor. -You can set this on the input binding on each processor. +In the case of the functional model, you can attach it to each function as a property. -For e.g. imagine that you have to two following functions. +For e.g. imagine that you have the following functions. ``` @Bean @@ -514,13 +514,32 @@ public java.util.function.Consumer> anotherProcess() { } ``` -Then you must set the application id for each, using the following binding properties. +Then you can set the application id for each, using the following binder level properties. -`spring.cloud.stream.kafka.streams.bindings.process_in.applicationId` +`spring.cloud.stream.kafka.streams.binder.process.applicationId` and -`spring.cloud.stream.kafka.streams.bindings.anotherProcess_in.applicationId` +`spring.cloud.stream.kafka.streams.binder.anotherProcess.applicationId` + +In the case of `StreamListener`, you need to set this on the first input binding on the processor. + +For e.g. imagine that you have to two following `StreamListener` based processors. + +``` +@StreamListener +public KStream process(@Input("input") > input) { + ... +} +``` + +Then you must set the application id for this using the following binding property. + +`spring.cloud.stream.kafka.streams.bindings.input.applicationId` + + +Fof function based model also, this approach of setting application id at the binding level will work. +However, setting per function at the binder level as we have seen above is much easier if you are using the functional model. For production deployments, it is highly recommended to explicitly specify the application ID through configuration. This is especially going to be very critical if you are auto scaling your application in which case you need to make sure that you are deploying each instance with the same application ID. @@ -533,9 +552,9 @@ In the case of `StreamListener`, instead of using the function bean name, the ge ====== Summary of setting Application ID -* Auto generated by the binder per processor in the application. This can be overridden by setting at the binding level such as `spring.cloud.stream.kafka.streams.bindings.process_in.applicationId`. -When you have more than one processor, then you have to choose one of these options - either fall back to the defaults or override per input binding. -* If you have a single processor, then you can use `spring.kafka.streams.applicationId`, `spring.application.name` or `spring.cloud.stream.binder.kafka.streams.applicationId`. +* Auto generated by the binder per processor in the application. This can be overridden by setting at the binding level such as `spring.cloud.stream.kafka.streams.bindings.process_in.applicationId` (or binder level per function in the case of functional model). +When you have more than one processor, then you have to choose one of these options - either fall back to the defaults or override. +* If you have a single processor, then you can use `spring.kafka.streams.applicationId`, `spring.application.name` or `spring.cloud.stream.kafka.streams.binder.applicationId`. ==== Custom bindings in the functional style @@ -596,13 +615,30 @@ Please note that this is a major change on default behavior from previous versio Kafka Streams binder will try to infer matching Serde types by looking at the type signature of `java.util.function.Function|Consumer` or `StreamListener`. Here is the order that it matches Serdes. - * First it looks at the types and see if they are one of the types exposed by Kafka Streams. If so, use them. +* If the application provides a bean of type `Serde` and if the return type is parameterized with the actual type of incoming key or value type, then it will use that `Serde` for inbound deserialization. +For e.g. if you have the following in the application, the binder detects that the incoming value type for the `KStream` matches with a type that is parameterized on a `Serde` bean. +It will use that for inbound deserialization. + + +``` +@Bean +public Serde, KStream> process() { +} +``` + + * Next, it looks at the types and see if they are one of the types exposed by Kafka Streams. If so, use them. Here are the Serde types that the binder will try to match from Kafka Streams. - Integer, Long, Short, Double, Float, byte[] and String. + Integer, Long, Short, Double, Float, byte[], UUID and String. * If none of the Serdes provided by Kafka Streams don't match the types, then it will use JsonSerde provided by Spring Kafka. In this case, the binder assumes that the types are JSON friendly. This is useful if you have multiple value objects as inputs since the binder will internally infer them to correct json Serde objects. Otherwise, you have to configure Serde and target types on them individually. + Before falling back to the `JsonSerde` though, the binder checks at the default Serdes's set at the Kafka Streams level to see if it is a Serde that it can match with the incoming KStream's types. If none of the above strategies worked, then the applications must provide the Serdes through configuration. This can be configured in two ways - binding or default. @@ -618,11 +654,11 @@ public BiFunction, KTable, KStre then, you can provide a binding level Serde using the following: ``` -spring.cloud.stream.kafka.streams.bindings.process_in_0.keySerde=CustomKeySerde -spring.cloud.stream.kafka.streams.bindings.process_in_0.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde +spring.cloud.stream.kafka.streams.bindings.process_in_0.consumer.keySerde=CustomKeySerde +spring.cloud.stream.kafka.streams.bindings.process_in_0.consumer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde -spring.cloud.stream.kafka.streams.bindings.process_in_1.keySerde=CustomKeySerde -spring.cloud.stream.kafka.streams.bindings.process_in_1.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde +spring.cloud.stream.kafka.streams.bindings.process_in_1.consumer.keySerde=CustomKeySerde +spring.cloud.stream.kafka.streams.bindings.process_in_1.consumer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde ``` If you want the default key/value Serdes to be used for inbound deserialization, you can do so at the binder level. @@ -635,7 +671,7 @@ spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde If you don't want the native decoding provided by Kafka, you can rely on the message conversion features that Spring Cloud Stream provides. Since native decoding is the default, in order to let Spring Cloud Stream deserialze the inbound value object, you need to explicitly disable native decoding. -For e.g. if you have the same BiFunction processor as above, then `spring.cloud.stream.bindings.process_in_0.nativeDecoding: false` +For e.g. if you have the same BiFunction processor as above, then `spring.cloud.stream.bindings.process_in_0.consumer.nativeDecoding: false` You need to disable native decoding for all the inputs individually. Otherwise, native decoding will still be applied for those you don't disable. By default, Spring Cloud Stream will use `application/json` as the content type and use an appropriate json message converter. @@ -650,34 +686,36 @@ Outbound serialization pretty much follows the same rules as above for inbound d As with the inbound deserialization, one major change from the previous versions of Spring Cloud Stream is that the serialization on the outbound is handled by Kafka natively. Before 3.0 versions of the binder, this was done by the framework itself. -Keys on the outbound are always serialized by Kafka using a matching Serde that is inferred by the binder. +Keys on the outbound are always serialized by Kafka using a matching `Serde` that is inferred by the binder. If it can't infer the type of the key, then that needs to be specified using configuration. Value serdes are inferred using the same rules used for inbound deserialization. -First it matches to see if the outbound type is of a Serde exposed by Kafka such as - Long, Short, Double, Float, byte[] and String. -If that doesnt't work, then fall back to JsonSerde provided by the Spring Kafka project. +First it matches to see if the outbound type is from a provided bean in the application. +If not, it checks to see if it matches with a `Serde` exposed by Kafka such as - Long, Short, Double, Float, byte[] and String. +If that doesnt't work, then fall back to JsonSerde provided by the Spring Kafka project, but first look at the default `Serde` configuration to see if there is a match. Keep in mind that all these happen transparently to the application. -If none of these work, then the user has to provide the Serde to use by configuration. +If none of these work, then the user has to provide the `Serde` to use by configuration. Lets say you are using the same `BiFunction` processor as above. Then you can configure outbound key/value Serdes as following. ``` -spring.cloud.stream.kafka.streams.bindings.process_out.keySerde=CustomKeySerde -spring.cloud.stream.kafka.streams.bindings.process_out.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde +spring.cloud.stream.kafka.streams.bindings.process_out.producer.keySerde=CustomKeySerde +spring.cloud.stream.kafka.streams.bindings.process_out.producer.valueSerde=io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde ``` -However, falling back to default Serdes for both input deserialization and output serialization is the last resort. -This may or may not work. Therefore, you need to ensure that you have a path forward for the application to correctly retrive the Serde. -If Serde inference fails, no binding level Serdes are provided, then the binder falls back to the default Serdes. +If Serde inference fails, and no binding level Serdes are provided, then the binder falls back to the default Serdes. `spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde` `spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde` +However, falling back to default Serdes for both input deserialization and output serialization is the last resort. +This may or may not work. Therefore, you need to ensure that you have a path forward for the application to correctly retrieve the Serde. + If your application uses the branching feature and has multiple output bindings, then these have to be configured per binding. Once again, if the binder is capable of inferring the Serde types, you don't need to do this configuration. If you don't want the native encoding provided by Kafka, but want to use the framework provided message conversion, then you need to explicitly disable native decoding since since native decoding is the default. -For e.g. if you have the same BiFunction processor as above, then `spring.cloud.stream.bindings.process_out.nativeEncoding: false` +For e.g. if you have the same BiFunction processor as above, then `spring.cloud.stream.bindings.process_out.producer.nativeEncoding: false` You need to disable native encoding for all the output individually in the case of branching. Otherwise, native encoding will still be applied for those you don't disable. By default, Spring Cloud Stream will use `application/json` as the content type and use an appropriate json message converter. diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index c5a3ce4ef..9c7a7e4e1 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Map; import java.util.Properties; -import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -41,6 +40,7 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.config.BindingProperties; @@ -54,6 +54,7 @@ import org.springframework.kafka.config.StreamsBuilderFactoryBean; import org.springframework.kafka.core.CleanupConfig; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; /** @@ -186,7 +187,8 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application @SuppressWarnings({"unchecked"}) protected StreamsBuilderFactoryBean buildStreamsBuilderAndRetrieveConfig(String beanNamePostPrefix, - ApplicationContext applicationContext, String inboundName) { + ApplicationContext applicationContext, String inboundName, + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { ConfigurableListableBeanFactory beanFactory = this.applicationContext .getBeanFactory(); @@ -200,16 +202,27 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application String bindingLevelApplicationId = extendedConsumerProperties.getApplicationId(); // override application.id if set at the individual binding level. + // We provide this for backward compatibility with StreamListener based processors. + // For function based processors see the next else if conditional block if (StringUtils.hasText(bindingLevelApplicationId)) { streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, bindingLevelApplicationId); } + else if (kafkaStreamsBinderConfigurationProperties != null && !CollectionUtils.isEmpty(kafkaStreamsBinderConfigurationProperties.getFunctions())) { + String applicationId = kafkaStreamsBinderConfigurationProperties.getFunctions().get(beanNamePostPrefix + ".applicationId"); + if (!StringUtils.isEmpty(applicationId)) { + streamConfigGlobalProperties.put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + } + } //If the application id is not set by any mechanism, then generate it. streamConfigGlobalProperties.computeIfAbsent(StreamsConfig.APPLICATION_ID_CONFIG, k -> { - String generatedApplicationID = beanNamePostPrefix + "-" + UUID.randomUUID().toString() + "-applicationId"; - LOG.info("Generated Kafka Streams Application ID: " + generatedApplicationID); + String generatedApplicationID = beanNamePostPrefix + "-applicationId"; + LOG.info("Binder Generated Kafka Streams Application ID: " + generatedApplicationID); + LOG.info("Use the binder generated application ID only for development and testing. "); + LOG.info("For production deployments, please consider explicitly setting an application ID using a configuration property."); + LOG.info("The generated applicationID is static and will be preserved over application restarts."); return generatedApplicationID; }); @@ -224,8 +237,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application Map kafkaStreamsDlqDispatchers = applicationContext .getBean("kafkaStreamsDlqDispatchers", Map.class); - KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration( - streamConfigGlobalProperties) { + KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties) { @Override public Properties asProperties() { Properties properties = super.asProperties(); @@ -248,6 +260,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition( "stream-builder-" + beanNamePostPrefix, streamsBuilderBeanDefinition); + //Removing the application ID from global properties so that the next function won't re-use it and cause race conditions. + streamConfigGlobalProperties.remove(StreamsConfig.APPLICATION_ID_CONFIG); + return applicationContext.getBean( "&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 96735101f..e0507c4fd 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -38,7 +38,6 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.BinderConfiguration; import org.springframework.cloud.stream.binder.kafka.streams.function.FunctionDetectorCondition; -import org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.streams.serde.CompositeNonNativeSerde; @@ -353,11 +352,11 @@ public class KafkaStreamsBinderSupportAutoConfiguration { KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate, ObjectProvider cleanupConfig, - KafkaStreamsBindableProxyFactory bindableProxyFactory, - StreamFunctionProperties streamFunctionProperties) { + StreamFunctionProperties streamFunctionProperties, + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { return new KafkaStreamsFunctionProcessor(bindingServiceProperties, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, kafkaStreamsBindingInformationCatalogue, kafkaStreamsMessageConversionDelegate, - cleanupConfig.getIfUnique(), bindableProxyFactory, streamFunctionProperties); + cleanupConfig.getIfUnique(), streamFunctionProperties, kafkaStreamsBinderConfigurationProperties); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index 1ebbf075d..65de5b1ce 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -48,6 +48,7 @@ import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.cloud.stream.binder.kafka.streams.function.KafkaStreamsBindableProxyFactory; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.cloud.stream.binding.StreamListenerErrorMessages; @@ -68,6 +69,7 @@ import org.springframework.util.StringUtils; public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderProcessor implements BeanFactoryAware { private static final Log LOG = LogFactory.getLog(KafkaStreamsFunctionProcessor.class); + private static final String OUTBOUND = "outbound"; private final BindingServiceProperties bindingServiceProperties; private final Map methodStreamsBuilderFactoryBeanMap = new HashMap<>(); @@ -76,13 +78,9 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro private final KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue; private final KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate; - private Set origInputs = new LinkedHashSet<>(); - private Set origOutputs = new LinkedHashSet<>(); - - private ResolvableType outboundResolvableType; - private KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory; private BeanFactory beanFactory; private StreamFunctionProperties streamFunctionProperties; + private KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties; public KafkaStreamsFunctionProcessor(BindingServiceProperties bindingServiceProperties, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, @@ -90,8 +88,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro KafkaStreamsBindingInformationCatalogue kafkaStreamsBindingInformationCatalogue, KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate, CleanupConfig cleanupConfig, - KafkaStreamsBindableProxyFactory bindableProxyFactory, - StreamFunctionProperties streamFunctionProperties) { + StreamFunctionProperties streamFunctionProperties, + KafkaStreamsBinderConfigurationProperties kafkaStreamsBinderConfigurationProperties) { super(bindingServiceProperties, kafkaStreamsBindingInformationCatalogue, kafkaStreamsExtendedBindingProperties, keyValueSerdeResolver, cleanupConfig); this.bindingServiceProperties = bindingServiceProperties; @@ -99,13 +97,12 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro this.keyValueSerdeResolver = keyValueSerdeResolver; this.kafkaStreamsBindingInformationCatalogue = kafkaStreamsBindingInformationCatalogue; this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate; - this.kafkaStreamsBindableProxyFactory = bindableProxyFactory; - this.origInputs.addAll(bindableProxyFactory.getInputs()); - this.origOutputs.addAll(bindableProxyFactory.getOutputs()); this.streamFunctionProperties = streamFunctionProperties; + this.kafkaStreamsBinderConfigurationProperties = kafkaStreamsBinderConfigurationProperties; } - private Map buildTypeMap(ResolvableType resolvableType) { + private Map buildTypeMap(ResolvableType resolvableType, + KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory) { Map resolvableTypeMap = new LinkedHashMap<>(); if (resolvableType != null && resolvableType.getRawClass() != null) { int inputCount = 1; @@ -119,13 +116,11 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro else { currentOutputGeneric = resolvableType.getGeneric(1); } - while (currentOutputGeneric != null && currentOutputGeneric.getRawClass() != null - && (functionOrConsumerFound(currentOutputGeneric))) { + while (currentOutputGeneric.getRawClass() != null && functionOrConsumerFound(currentOutputGeneric)) { inputCount++; currentOutputGeneric = currentOutputGeneric.getGeneric(1); } - - final Set inputs = new LinkedHashSet<>(origInputs); + final Set inputs = new LinkedHashSet<>(kafkaStreamsBindableProxyFactory.getInputs()); final Iterator iterator = inputs.iterator(); @@ -134,6 +129,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro ResolvableType iterableResType = resolvableType; int i = resolvableType.getRawClass().isAssignableFrom(BiFunction.class) || resolvableType.getRawClass().isAssignableFrom(BiConsumer.class) ? 2 : 1; + ResolvableType outboundResolvableType; if (i == inputCount) { outboundResolvableType = iterableResType.getGeneric(i); } @@ -148,6 +144,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro } outboundResolvableType = iterableResType.getGeneric(1); } + resolvableTypeMap.put(OUTBOUND, outboundResolvableType); } return resolvableTypeMap; } @@ -157,7 +154,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro iterableResType.getRawClass().equals(Consumer.class); } - private void popuateResolvableTypeMap(ResolvableType resolvableType, Map resolvableTypeMap, Iterator iterator) { + private void popuateResolvableTypeMap(ResolvableType resolvableType, Map resolvableTypeMap, + Iterator iterator) { final String next = iterator.next(); resolvableTypeMap.put(next, resolvableType.getGeneric(0)); if (resolvableType.getRawClass() != null && @@ -166,36 +164,41 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro && iterator.hasNext()) { resolvableTypeMap.put(iterator.next(), resolvableType.getGeneric(1)); } - origInputs.remove(next); } + /** + * This method must be kept stateless. In the case of multiple function beans in an application, + * isolated {@link KafkaStreamsBindableProxyFactory} instances are passed in separately for those functions. If the + * state is shared between invocations, that will create potential race conditions. Hence, invocations of this method + * should not be dependent on state modified by a previous invocation. + * + * @param resolvableType type of the binding + * @param functionName bean name of the function + * @param kafkaStreamsBindableProxyFactory bindable proxy factory for the Kafka Streams type + */ @SuppressWarnings({ "unchecked", "rawtypes" }) - public void setupFunctionInvokerForKafkaStreams(ResolvableType resolvableType, String functionName) { - final Map stringResolvableTypeMap = buildTypeMap(resolvableType); + public void setupFunctionInvokerForKafkaStreams(ResolvableType resolvableType, String functionName, + KafkaStreamsBindableProxyFactory kafkaStreamsBindableProxyFactory) { + final Map stringResolvableTypeMap = buildTypeMap(resolvableType, kafkaStreamsBindableProxyFactory); + ResolvableType outboundResolvableType = stringResolvableTypeMap.remove(OUTBOUND); Object[] adaptedInboundArguments = adaptAndRetrieveInboundArguments(stringResolvableTypeMap, functionName); try { if (resolvableType.getRawClass() != null && resolvableType.getRawClass().equals(Consumer.class)) { Consumer consumer = (Consumer) this.beanFactory.getBean(functionName); - Assert.isTrue(consumer != null, - "No corresponding consumer beans found in the catalog"); consumer.accept(adaptedInboundArguments[0]); } else if (resolvableType.getRawClass() != null && resolvableType.getRawClass().equals(BiConsumer.class)) { BiConsumer biConsumer = (BiConsumer) this.beanFactory.getBean(functionName); - Assert.isTrue(biConsumer != null, - "No corresponding biConsumer beans found"); biConsumer.accept(adaptedInboundArguments[0], adaptedInboundArguments[1]); } else { Object result; if (resolvableType.getRawClass() != null && resolvableType.getRawClass().equals(BiFunction.class)) { BiFunction biFunction = (BiFunction) beanFactory.getBean(functionName); - Assert.isTrue(biFunction != null, "Biunction bean cannot be null"); result = biFunction.apply(adaptedInboundArguments[0], adaptedInboundArguments[1]); } else { Function function = (Function) beanFactory.getBean(functionName); - Assert.isTrue(function != null, "Function bean cannot be null"); result = function.apply(adaptedInboundArguments[0]); } int i = 1; @@ -212,11 +215,11 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro if (result != null) { kafkaStreamsBindingInformationCatalogue.setOutboundKStreamResolvable( outboundResolvableType != null ? outboundResolvableType : resolvableType.getGeneric(1)); - final Set outputs = new TreeSet<>(origOutputs); + final Set outputs = new TreeSet<>(kafkaStreamsBindableProxyFactory.getOutputs()); final Iterator outboundDefinitionIterator = outputs.iterator(); if (result.getClass().isArray()) { - // Binding target as the output bindings were deffered in the KafkaStreamsBindableProxyFacotyr + // Binding target as the output bindings were deferred in the KafkaStreamsBindableProxyFactory // due to the fact that it didn't know the returned array size. At this point in the execution, // we know exactly the number of outbound components (from the array length), so do the binding. final int length = ((Object[]) result).length; @@ -229,7 +232,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro for (int ij = 0; ij < length; ij++) { String next = iterator.next(); - this.kafkaStreamsBindableProxyFactory.addOutputBinding(next, KStream.class); + kafkaStreamsBindableProxyFactory.addOutputBinding(next, KStream.class); RootBeanDefinition rootBeanDefinition1 = new RootBeanDefinition(); rootBeanDefinition1.setInstanceSupplier(() -> kafkaStreamsBindableProxyFactory.getOutputHolders().get(next).getBoundTarget()); registry.registerBeanDefinition(next, rootBeanDefinition1); @@ -246,8 +249,6 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro if (outboundDefinitionIterator.hasNext()) { final String next = outboundDefinitionIterator.next(); Object targetBean = this.applicationContext.getBean(next); - this.origOutputs.remove(next); - KStreamBoundElementFactory.KStreamWrapper boundElement = (KStreamBoundElementFactory.KStreamWrapper) targetBean; boundElement.wrap((KStream) result); @@ -291,7 +292,7 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro //Retrieve the StreamsConfig created for this method if available. //Otherwise, create the StreamsBuilderFactory and get the underlying config. if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(functionName)) { - StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(functionName, applicationContext, input); + StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(functionName, applicationContext, input, kafkaStreamsBinderConfigurationProperties); this.methodStreamsBuilderFactoryBeanMap.put(functionName, streamsBuilderFactoryBean); } try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index 8e8461ad0..e471fa361 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -246,7 +246,7 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator extends AbstractKafkaStr if (!this.methodStreamsBuilderFactoryBeanMap.containsKey(method)) { StreamsBuilderFactoryBean streamsBuilderFactoryBean = buildStreamsBuilderAndRetrieveConfig(method.getDeclaringClass().getSimpleName() + "-" + method.getName(), applicationContext, - inboundName); + inboundName, null); this.methodStreamsBuilderFactoryBeanMap.put(method, streamsBuilderFactoryBean); } try { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java index f3532080b..e44617c98 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KeyValueSerdeResolver.java @@ -16,7 +16,11 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.lang.reflect.Method; +import java.util.Arrays; import java.util.Map; +import java.util.Optional; +import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -27,13 +31,19 @@ import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.ResolvableType; import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; /** @@ -60,7 +70,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Lei Chen */ -public class KeyValueSerdeResolver { +public class KeyValueSerdeResolver implements ApplicationContextAware { private static final Log LOG = LogFactory.getLog(KeyValueSerdeResolver.class); @@ -68,6 +78,8 @@ public class KeyValueSerdeResolver { private final KafkaStreamsBinderConfigurationProperties binderConfigurationProperties; + private ConfigurableApplicationContext context; + KeyValueSerdeResolver(Map streamConfigGlobalProperties, KafkaStreamsBinderConfigurationProperties binderConfigurationProperties) { this.streamConfigGlobalProperties = streamConfigGlobalProperties; @@ -252,10 +264,11 @@ public class KeyValueSerdeResolver { if (resolvableType != null && (isResolvalbeKafkaStreamsType(resolvableType) || isResolvableKStreamArrayType(resolvableType))) { ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(0) : resolvableType.getGeneric(0); - keySerde = getSerde(generic); + Serde fallbackSerde = getFallbackSerde("default.key.serde"); + keySerde = getSerde(generic, fallbackSerde); } if (keySerde == null) { - keySerde = getFallbackSerde("default.key.serde"); + keySerde = Serdes.ByteArray(); } } keySerde.configure(this.streamConfigGlobalProperties, true); @@ -276,40 +289,112 @@ public class KeyValueSerdeResolver { GlobalKTable.class.isAssignableFrom(resolvableType.getRawClass())); } - private Serde getSerde(ResolvableType generic) { + private Serde getSerde(ResolvableType generic, Serde fallbackSerde) { Serde serde = null; - if (generic.getRawClass() != null) { - if (Integer.class.isAssignableFrom(generic.getRawClass())) { + + Map beansOfType = context.getBeansOfType(Serde.class); + Serde[] serdeBeans = new Serde[1]; + + final Class genericRawClazz = generic.getRawClass(); + beansOfType.forEach((k, v) -> { + final Class classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition) + context.getBeanFactory().getBeanDefinition(k)) + .getMetadata().getClassName(), + ClassUtils.getDefaultClassLoader()); + try { + Method[] methods = classObj.getMethods(); + Optional serdeBeanMethod = Arrays.stream(methods).filter(m -> m.getName().equals(k)).findFirst(); + if (serdeBeanMethod.isPresent()) { + Method method = serdeBeanMethod.get(); + ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj); + ResolvableType serdeBeanGeneric = resolvableType.getGeneric(0); + Class serdeGenericRawClazz = serdeBeanGeneric.getRawClass(); + if (serdeGenericRawClazz != null && genericRawClazz != null) { + if (serdeGenericRawClazz.isAssignableFrom(genericRawClazz)) { + serdeBeans[0] = v; + } + } + } + } + catch (Exception e) { + // Pass through... + } + + }); + + if (serdeBeans[0] != null) { + return serdeBeans[0]; + } + + if (genericRawClazz != null) { + if (Integer.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.Integer(); } - else if (Long.class.isAssignableFrom(generic.getRawClass())) { + else if (Long.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.Long(); } - else if (Short.class.isAssignableFrom(generic.getRawClass())) { + else if (Short.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.Short(); } - else if (Double.class.isAssignableFrom(generic.getRawClass())) { + else if (Double.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.Double(); } - else if (Float.class.isAssignableFrom(generic.getRawClass())) { + else if (Float.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.Float(); } - else if (byte[].class.isAssignableFrom(generic.getRawClass())) { + else if (byte[].class.isAssignableFrom(genericRawClazz)) { serde = Serdes.ByteArray(); } - else if (String.class.isAssignableFrom(generic.getRawClass())) { + else if (String.class.isAssignableFrom(genericRawClazz)) { serde = Serdes.String(); } + else if (UUID.class.isAssignableFrom(genericRawClazz)) { + serde = Serdes.UUID(); + } + else if (!isSerdeFromStandardDefaults(fallbackSerde)) { + //User purposely set a default serde that is not one of the above + serde = fallbackSerde; + } else { // If the type is Object, then skip assigning the JsonSerde and let the fallback mechanism takes precedence. - if (!generic.getRawClass().isAssignableFrom((Object.class))) { - serde = new JsonSerde(generic.getRawClass()); + if (!genericRawClazz.isAssignableFrom((Object.class))) { + serde = new JsonSerde(genericRawClazz); } } } return serde; } + private boolean isSerdeFromStandardDefaults(Serde serde) { + if (serde != null) { + if (Serdes.Integer().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.Long().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.Short().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.Double().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.Float().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.ByteArray().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.String().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + else if (Serdes.UUID().getClass().isAssignableFrom(serde.getClass())) { + return true; + } + } + return false; + } + private Serde getValueSerde(String valueSerdeString) throws ClassNotFoundException { @@ -343,16 +428,20 @@ public class KeyValueSerdeResolver { if (resolvableType != null && ((isResolvalbeKafkaStreamsType(resolvableType)) || (isResolvableKStreamArrayType(resolvableType)))) { + Serde fallbackSerde = getFallbackSerde("default.value.serde"); ResolvableType generic = resolvableType.isArray() ? resolvableType.getComponentType().getGeneric(1) : resolvableType.getGeneric(1); - valueSerde = getSerde(generic); + valueSerde = getSerde(generic, fallbackSerde); } - if (valueSerde == null) { - valueSerde = getFallbackSerde("default.value.serde"); + valueSerde = Serdes.ByteArray(); } } return valueSerde; } + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + context = (ConfigurableApplicationContext) applicationContext; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java index 037b8397f..09696860d 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBindableProxyFactory.java @@ -241,6 +241,10 @@ public class KafkaStreamsBindableProxyFactory extends AbstractBindableProxyFacto .createOutput(output), true)); } + public String getFunctionName() { + return functionName; + } + public Map getOutputHolders() { return outputHolders; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java index af29eb41a..075e749a9 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java @@ -41,9 +41,10 @@ public class KafkaStreamsFunctionAutoConfiguration { @Conditional(FunctionDetectorCondition.class) public KafkaStreamsFunctionProcessorInvoker kafkaStreamsFunctionProcessorInvoker( KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor, - KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor) { + KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, + KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories) { return new KafkaStreamsFunctionProcessorInvoker(kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes(), - kafkaStreamsFunctionProcessor); + kafkaStreamsFunctionProcessor, kafkaStreamsBindableProxyFactories); } @Bean @@ -65,7 +66,7 @@ public class KafkaStreamsFunctionAutoConfiguration { .addGenericArgumentValue(kafkaStreamsFunctionBeanPostProcessor.getResolvableTypes().get(s)); rootBeanDefinition.getConstructorArgumentValues() .addGenericArgumentValue(s); - registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory", rootBeanDefinition); + registry.registerBeanDefinition("kafkaStreamsBindableProxyFactory-" + s, rootBeanDefinition); } }; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java index 77bb18367..9f89b42b6 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; +import java.util.Arrays; import java.util.Map; +import java.util.Optional; import javax.annotation.PostConstruct; @@ -32,16 +34,22 @@ public class KafkaStreamsFunctionProcessorInvoker { private final KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor; private final Map resolvableTypeMap; + private final KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories; public KafkaStreamsFunctionProcessorInvoker(Map resolvableTypeMap, - KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor) { + KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, + KafkaStreamsBindableProxyFactory[] kafkaStreamsBindableProxyFactories) { this.kafkaStreamsFunctionProcessor = kafkaStreamsFunctionProcessor; this.resolvableTypeMap = resolvableTypeMap; + this.kafkaStreamsBindableProxyFactories = kafkaStreamsBindableProxyFactories; } @PostConstruct void invoke() { - resolvableTypeMap.forEach((key, value) -> - this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key)); + resolvableTypeMap.forEach((key, value) -> { + Optional proxyFactory = + Arrays.stream(kafkaStreamsBindableProxyFactories).filter(p -> p.getFunctionName().equals(key)).findFirst(); + this.kafkaStreamsFunctionProcessor.setupFunctionInvokerForKafkaStreams(value, key, proxyFactory.get()); + }); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java index 0f2f577d7..48a618b4e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java @@ -16,6 +16,9 @@ package org.springframework.cloud.stream.binder.kafka.streams.properties; +import java.util.HashMap; +import java.util.Map; + import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; @@ -54,6 +57,8 @@ public class KafkaStreamsBinderConfigurationProperties private String applicationId; + private Map functions = new HashMap<>(); + private StateStoreRetry stateStoreRetry = new StateStoreRetry(); public StateStoreRetry getStateStoreRetry() { @@ -64,6 +69,14 @@ public class KafkaStreamsBinderConfigurationProperties this.stateStoreRetry = stateStoreRetry; } + public Map getFunctions() { + return functions; + } + + public void setFunctions(Map functions) { + this.functions = functions; + } + public String getApplicationId() { return this.applicationId; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java new file mode 100644 index 000000000..6b3e5ef8a --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/MultipleFunctionsInSameAppTests.java @@ -0,0 +1,140 @@ +/* + * Copyright 2019-2019 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.function; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import java.util.function.Function; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.util.Assert; + +import static org.assertj.core.api.Assertions.assertThat; + +public class MultipleFunctionsInSameAppTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "coffee", "electronics"); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + private static Consumer consumer; + + private static CountDownLatch countDownLatch = new CountDownLatch(2); + + @BeforeClass + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "coffee", "electronics"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamWordCountFunction() throws InterruptedException { + SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext ignored = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.process_in.destination=purchases", + "--spring.cloud.stream.bindings.process_out_0.destination=coffee", + "--spring.cloud.stream.bindings.process_out_1.destination=electronics", + "--spring.cloud.stream.bindings.analyze_in_0.destination=coffee", + "--spring.cloud.stream.bindings.analyze_in_1.destination=electronics", + "--spring.cloud.stream.kafka.streams.binder.functions.analyze.applicationId=analyze-id-0", + "--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-0", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + receiveAndValidate("purchases", "coffee", "electronics"); + } + } + + private void receiveAndValidate(String in, String... out) throws InterruptedException { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic(in); + template.sendDefault("coffee"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, out[0]); + assertThat(cr.value().contains("coffee")).isTrue(); + + template.sendDefault("electronics"); + cr = KafkaTestUtils.getSingleRecord(consumer, out[1]); + assertThat(cr.value().contains("electronics")).isTrue(); + + Assert.isTrue(countDownLatch.await(5, TimeUnit.SECONDS), "Analyze (BiConsumer) method didn't receive all the expected records"); + } + finally { + pf.destroy(); + } + } + + @EnableAutoConfiguration + public static class MultipleFunctionsInSameApp { + + @Bean + public Function, KStream[]> process() { + return input -> input.branch( + (s, p) -> p.equalsIgnoreCase("coffee"), + (s, p) -> p.equalsIgnoreCase("electronics")); + } + + @Bean + public BiConsumer, KStream> analyze() { + return (coffee, electronics) -> { + coffee.foreach((s, p) -> countDownLatch.countDown()); + electronics.foreach((s, p) -> countDownLatch.countDown()); + }; + } + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java new file mode 100644 index 000000000..bd43744a1 --- /dev/null +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java @@ -0,0 +1,121 @@ +/* + * Copyright 2019-2019 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.function; + +import java.lang.reflect.Method; +import java.util.Date; +import java.util.function.Function; + +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serializer; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolver; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.core.ResolvableType; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.util.Assert; + +public class SerdesProvidedAsBeansTests { + + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + + @Test + public void testKstreamWordCountFunction() throws NoSuchMethodException { + SpringApplication app = new SpringApplication(SerdeProvidedAsBeanApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.process_in.destination=purchases", + "--spring.cloud.stream.bindings.process_out.destination=coffee", + "--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-0", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" + + "=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + final Method method = SerdeProvidedAsBeanApp.class.getMethod("process"); + + ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, SerdeProvidedAsBeanApp.class); + + final KeyValueSerdeResolver keyValueSerdeResolver = context.getBean(KeyValueSerdeResolver.class); + final BindingServiceProperties bindingServiceProperties = context.getBean(BindingServiceProperties.class); + final KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = context.getBean(KafkaStreamsExtendedBindingProperties.class); + + final ConsumerProperties consumerProperties = bindingServiceProperties.getBindingProperties("process_in").getConsumer(); + final KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties = kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("process_in"); + kafkaStreamsExtendedBindingProperties.getExtendedConsumerProperties("process_in"); + final Serde inboundValueSerde = keyValueSerdeResolver.getInboundValueSerde(consumerProperties, kafkaStreamsConsumerProperties, resolvableType.getGeneric(0)); + + Assert.isTrue(inboundValueSerde instanceof FooSerde, "Inbound Value Serde is not matched"); + + final ProducerProperties producerProperties = bindingServiceProperties.getBindingProperties("process_out").getProducer(); + final KafkaStreamsProducerProperties kafkaStreamsProducerProperties = kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("process_out"); + kafkaStreamsExtendedBindingProperties.getExtendedProducerProperties("process_out"); + final Serde outboundValueSerde = keyValueSerdeResolver.getOutboundValueSerde(producerProperties, kafkaStreamsProducerProperties, resolvableType.getGeneric(1)); + + Assert.isTrue(outboundValueSerde instanceof FooSerde, "Outbound Value Serde is not matched"); + } + } + + static class FooSerde implements Serde { + @Override + public Serializer serializer() { + return null; + } + + @Override + public Deserializer deserializer() { + return null; + } + } + + @EnableAutoConfiguration + public static class SerdeProvidedAsBeanApp { + + @Bean + public Function, KStream> process() { + return input -> input; + } + + @Bean + public Serde fooSerde() { + return new FooSerde<>(); + } + } +}