diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 971623ce8..c246ab446 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -524,7 +524,8 @@ Spring Cloud Stream Kafka support also includes a binder specifically designed f Using this binder, applications can be written that leverage the Kafka Streams API. For more information on Kafka Streams, see https://kafka.apache.org/documentation/streams/developer-guide[Kafka Streams API Developer Manual] -Kafka Streams support in Spring Cloud Stream is based on the foundations provided by the Spring Kafka project. For details on that support, see http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafaka Streams Support in Spring Kafka]. +Kafka Streams support in Spring Cloud Stream is based on the foundations provided by the Spring Kafka project. +For details on that support, see http://docs.spring.io/spring-kafka/reference/html/_reference.html#kafka-streams[Kafaka Streams Support in Spring Kafka]. Here are the maven coordinates for the Spring Cloud Stream KStream binder artifact. @@ -536,8 +537,8 @@ Here are the maven coordinates for the Spring Cloud Stream KStream binder artifa ---- -In addition to leveraging the Spring Cloud Stream programming model which is based on Spring Boot, one of the main other benefits that the KStream binder provides is the fact that it avoids the boilerplate configuration that one needs to write when using the Kafka Streams API directly. -High level streams DSL provided through the Kafka Streams API can be used through Spring Cloud Stream in the current support. +High level streams DSL provided through the Kafka Streams API can be used through Spring Cloud Stream support. +Kafka Streams applications using the Spring Cloud Stream support can only be written using the processor model, i.e. messages read from an inbound topic and messages written to an outbound topic. === Usage example of high level streams DSL @@ -553,36 +554,134 @@ public class WordCountProcessorApplication { @SendTo("output") public KStream process(KStream input) { return input - .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) - .map((key, word) -> new KeyValue<>(word, word)) - .groupByKey(Serdes.String(), Serdes.String()) - .count(TimeWindows.of(5000), "store-name") - .toStream() - .map((w, c) -> new KeyValue<>(null, "Count for " + w.key() + ": " + c)); - } + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .groupBy((key, value) -> value) + .windowedBy(TimeWindows.of(5000)) + .count(Materialized.as("WordCounts-multi")) + .toStream() + .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); + } public static void main(String[] args) { SpringApplication.run(WordCountProcessorApplication.class, args); } ---- -If you build it as Spring Boot runnable fat jar, you can run the above example in the following way: +If you build it as a Spring Boot uber jar, you can run the above example in the following way: [source] ---- java -jar uber.jar --spring.cloud.stream.bindings.input.destination=words --spring.cloud.stream.bindings.output.destination=counts ---- -This means that the application will listen from the incoming Kafka topic words and write to the output topic counts. +This means that the application will listen from the incoming Kafka topic `words` and write to the output topic `counts`. Spring Cloud Stream will ensure that the messages from both the incoming and outgoing topics are bound as KStream objects. -As one may observe, the developer can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor rather than setting up the streams specific configuration required by the Kafka Streams infrastructure. -All those boilerplate is handled by Spring Cloud Stream behind the scenes. +Applications can exclusively focus on the business aspects of the code, i.e. writing the logic required in the processor rather than setting up the streams specific configuration required by the Kafka Streams infrastructure. +All such interactions are handled by the framework. + +=== Message conversion in Spring Cloud Stream Kafka Streams applications + +If the following property is set (default is false), the framework skips all message conversions on the outbound (producer) side and it is then done by Kafka itself. + +`spring.cloud.stream.bindings.output.producer.useNativeEncoding`. + +Similarly, if the following property is set (default is false), any message conversion is skipped on the inbound (consumer) side and natively done by Kafka. + +`spring.cloud.stream.bindings.input.consumer.useNativeDecoding`. + +When native encoding is disabled, then the messages on the outbound are converted by Spring Cloud Stream using the provided contentType. +If no contentType is set by the application, it defaults to `application/json`. + +By default, all the out of the box message converters, serialize the data as `byte[]` encoding the proper contentType. +In most situations, this is what you want to do, but if other formats than `byte[]` are desired, then ann appropriate message converter needs to be registered in the context and corresponding contentType specified as a property. +When doing this way, Serdes should be overridden on the producer using the following property. + +`spring.cloud.stream.kstream.bindings.output.producer.valueSerde`. + +Keys will not get converted, but if the Serdes are different for keys from what is given as the common Serde, you can override that using the following property. + +`spring.cloud.stream.kstream.bindings.output.producer.keySerde`. + +=== Support for branching in Kafka Streams API + +Kafka Streams allow outbound data to be split into multiple topics based on some predicates. +Spring Cloud Stream Kafka Streams binder provides support for this feature without losing the overall programming model exposed through `StreamListener` in the end user application. +You write the application in the usual way as demonstrated above in the word count example. +The actual splitting and branching into multiple topics are done by the framework behind the scenes. +When using the branching feature, you are required to do two things. +First, you need to provide the following property that specifies the extra branches (topics) in the order. +The first topic will always be the one specified through the main outbound destination. + +`spring.cloud.stream.kstream.bindings.output.producer.additionalBranches=foo,bar` + +If your main output destination is foobar provided through `spring.cloud.stream.bindings.output.destination=foobar`, then your 3 output branches (topics) are foobar, foo and bar. + +Second, you need to provide a `Bean` in your application context, that returns a `org.apache.kafka.streams.kstream.Predicate[]`. +The presence of this bean is the trigger to the framework to perform branching into multiple topics. +The individual Predicates in this bean, should match with the output branches in the same order. +Each Predicate should get its own output branch, otherwise, it fails. +If you provide more output branches than there are Predicates, that is fine, but the number of branches cannot be less than the Predicates. + +Here is an example: + +[source] +---- +@EnableBinding(KStreamProcessor.class) +@EnableAutoConfiguration +public static class WordCountProcessorApplication { + + @Autowired + private TimeWindows timeWindows; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .groupBy((key, value) -> value) + .windowedBy(timeWindows) + .count(Materialized.as("WordCounts-1")) + .toStream() + .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); + } + + @Bean + public Predicate[] predicates() { + Predicate isEnglish = (k, v) -> v.word.equals("english"); + Predicate isFrench = (k, v) -> v.word.equals("french"); + Predicate isSpanish = (k, v) -> v.word.equals("spanish"); + return new Predicate[] {isEnglish, isFrench, isSpanish}; + } + +} +---- + +Then in the properties: + +[source] +---- +spring.cloud.stream.bindings.output.contentType: application/json +spring.cloud.stream.kstream.binder.configuration.commit.interval.ms: 1000 +spring.cloud.stream.kstream.binder.configuration: + key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde + value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde +spring.cloud.stream.bindings.output: + destination: foobar + producer: + headerMode: raw +spring.cloud.stream.kstream.bindings.output.producer.additionalBranches: foo,bar +spring.cloud.stream.bindings.input: + destination: words + consumer: + headerMode: raw +---- === Support for interactive queries If access to the `KafkaStreams` is needed for interactive queries, the internal `KafkaStreams` instance can be accessed via `KStreamBuilderFactoryBean.getKafkaStreams()`. -You can autowire the `KStreamBuilderFactoryBean` instance provided by the KStream binder. Then you can get `KafkaStreams` instance from it and retrieve the underlying store, execute queries on it, etc. +You can autowire the `KStreamBuilderFactoryBean` instance provided by the KStream binder. +Then you get `KafkaStreams` instance from it and retrieve the underlying store, execute queries on it, etc. === Kafka Streams properties @@ -599,7 +698,7 @@ spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.co spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000 ---- - For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs. +For more information about all the properties that may go into streams configuration, see StreamsConfig JavaDocs. There can also be binding specific properties. @@ -611,6 +710,24 @@ spring.cloud.stream.kstream.bindings.output.producer.keySerde=org.apache.kafka.c spring.cloud.stream.kstream.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde ---- +Additional output branches: + +[source] +---- +spring.cloud.stream.kstream.bindings.output.producer.additionalBranches (comma separated values) +---- + +TimeWindow properties: + +[source] +---- +spring.cloud.stream.kstream.timeWindow.length (milliseconds) + +When this property is given, you can autowire a `TimeWindows` bean into the application. + +spring.cloud.stream.kstream.timeWindow.advanceBy (milliseconds) +---- + [[kafka-error-channels]] == Error Channels diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java index 7a5253f33..4cbb680e0 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java @@ -23,6 +23,8 @@ import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KeyValueMapper; +import org.apache.kafka.streams.kstream.Predicate; import org.apache.kafka.streams.kstream.Produced; import org.springframework.cloud.stream.binder.AbstractBinder; @@ -38,7 +40,7 @@ import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProv import org.springframework.cloud.stream.binder.kstream.config.KStreamConsumerProperties; import org.springframework.cloud.stream.binder.kstream.config.KStreamExtendedBindingProperties; import org.springframework.cloud.stream.binder.kstream.config.KStreamProducerProperties; -import org.springframework.messaging.Message; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** @@ -57,20 +59,27 @@ public class KStreamBinder extends private final KafkaBinderConfigurationProperties binderConfigurationProperties; + private Predicate[] predicates; + + private final MessageConversionDelegate messageConversionDelegate; + public KStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, - KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig, + MessageConversionDelegate messageConversionDelegate) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kStreamExtendedBindingProperties = kStreamExtendedBindingProperties; this.streamsConfig = streamsConfig; + this.messageConversionDelegate = messageConversionDelegate; } @Override protected Binding> doBindConsumer(String name, String group, - KStream inputTarget, ExtendedConsumerProperties properties) { + KStream inputTarget, + ExtendedConsumerProperties properties) { - ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties( + ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( new KafkaConsumerProperties()); this.kafkaTopicProvisioner.provisionConsumerDestination(name, group, extendedConsumerProperties); return new DefaultBinding<>(name, group, inputTarget, null); @@ -80,19 +89,43 @@ public class KStreamBinder extends @SuppressWarnings("unchecked") protected Binding> doBindProducer(String name, KStream outboundBindTarget, ExtendedProducerProperties properties) { - ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties( + ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>( new KafkaProducerProperties()); this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties); - outboundBindTarget = outboundBindTarget - .map((k, v) -> KeyValue.pair(k, ((Message) v).getPayload())); - Serde keySerde = Serdes.ByteArray(); - Serde valueSerde = Serdes.ByteArray(); - if (properties.isUseNativeEncoding()) { - outboundBindTarget.to(name, Produced.with((Serde) keySerde, (Serde) valueSerde)); + String[] branches = new String[]{}; + if (predicates != null && predicates.length > 0) { + String additionalBranches = properties.getExtension().getAdditionalBranches(); + if (!StringUtils.hasText(additionalBranches)) { + Assert.isTrue(predicates.length == 1, "More than 1 predicate bean found, but no additional output branches"); + } + else { + branches = StringUtils.commaDelimitedListToStringArray(additionalBranches); + Assert.isTrue(branches.length + 1 >= predicates.length, + "Number of output topics and org.apache.kafka.streams.kstream.Predicate[] beans don't match"); + for (String branch : branches) { + this.kafkaTopicProvisioner.provisionProducerDestination(branch, extendedProducerProperties); + } + } } - else { - try { + + Serde keySerde = getKeySerde(properties); + Serde valueSerde = getValueSerde(properties); + + to(properties.isUseNativeEncoding(), name, outboundBindTarget, (Serde) keySerde, (Serde) valueSerde, branches); + + return new DefaultBinding<>(name, null, outboundBindTarget, null); + } + + private Serde getKeySerde(ExtendedProducerProperties properties) { + Serde keySerde; + try { + if (properties.isUseNativeEncoding()) { + keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray(); + + } + else { if (StringUtils.hasText(properties.getExtension().getKeySerde())) { keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class); if (keySerde instanceof Configurable) { @@ -103,6 +136,23 @@ public class KStreamBinder extends keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ? Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray(); } + } + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + return keySerde; + } + + private Serde getValueSerde(ExtendedProducerProperties properties) { + Serde valueSerde; + try { + if (properties.isUseNativeEncoding()) { + valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ? + Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray(); + + } + else { if (StringUtils.hasText(properties.getExtension().getValueSerde())) { valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class); @@ -110,12 +160,53 @@ public class KStreamBinder extends ((Configurable) valueSerde).configure(streamsConfig.originals()); } } - outboundBindTarget.to(name, Produced.with((Serde) keySerde, (Serde) valueSerde)); - } catch (ClassNotFoundException e) { - throw new IllegalStateException("Serde class not found: ", e); + else { + valueSerde = Serdes.ByteArray(); + } } } - return new DefaultBinding<>(name, null, outboundBindTarget, null); + catch (ClassNotFoundException e) { + throw new IllegalStateException("Serde class not found: ", e); + } + return valueSerde; + } + + @SuppressWarnings("unchecked") + private void to(boolean isNativeEncoding, String name, KStream outboundBindTarget, + Serde keySerde, Serde valueSerde, String[] branches) { + KeyValueMapper> keyValueMapper = null; + if (!isNativeEncoding) { + keyValueMapper = messageConversionDelegate.outboundKeyValueMapper(name); + } + if (predicates != null && predicates.length > 0) { + KStream[] toBranches = outboundBindTarget.branch(predicates); + String[] topics = getOutputTopicsInProperOrder(name, branches); + for (int i = 0; i < toBranches.length; i++) { + if (!isNativeEncoding) { + toBranches[i].map(keyValueMapper).to(topics[i], Produced.with(keySerde, valueSerde)); + } + else { + toBranches[i].to(topics[i], Produced.with(keySerde, valueSerde)); + } + } + } else { + if (!isNativeEncoding) { + outboundBindTarget.map(keyValueMapper).to(name, Produced.with(keySerde, valueSerde)); + } + else { + outboundBindTarget.to(name, Produced.with(keySerde, valueSerde)); + } + } + } + + private static String[] getOutputTopicsInProperOrder(String name, String[] branches) { + String[] topics = new String[branches.length + 1]; + topics[0] = name; + int j = 1; + for (String branch : branches) { + topics[j++] = branch; + } + return topics; } @Override @@ -128,4 +219,8 @@ public class KStreamBinder extends return this.kStreamExtendedBindingProperties.getExtendedProducerProperties(channelName); } + public void setPredicates(Predicate[] predicates) { + this.predicates = predicates; + } + } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java index 2b53156c2..c383c098b 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBoundElementFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -16,28 +16,20 @@ package org.springframework.cloud.stream.binder.kstream; -import java.util.HashMap; -import java.util.Map; - import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KeyValueMapper; import org.springframework.aop.framework.ProxyFactory; -import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binding.AbstractBindingTargetFactory; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.MimeType; import org.springframework.util.StringUtils; /** @@ -50,14 +42,10 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory keyValue; BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); String contentType = bindingProperties.getContentType(); - if (!StringUtils.isEmpty(contentType)) { + if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) { Message message = MessageBuilder.withPayload(value) .setHeader(MessageHeaders.CONTENT_TYPE, contentType).build(); keyValue = new KeyValue<>(key, message); @@ -83,13 +71,9 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory delegate; - private final MessageConverter messageConverter; - private final BindingServiceProperties bindingServiceProperties; - private String name; - - KStreamWrapperHandler(MessageConverter messageConverter, - BindingServiceProperties bindingServiceProperties, - String name) { - this.messageConverter = messageConverter; - this.bindingServiceProperties = bindingServiceProperties; - this.name = name; - } - public void wrap(KStream delegate) { Assert.notNull(delegate, "delegate cannot be null"); Assert.isNull(this.delegate, "delegate already set to " + this.delegate); - ProducerProperties producer = bindingServiceProperties.getBindingProperties(name).getProducer(); - - if (messageConverter != null && !producer.isUseNativeEncoding()) { - KeyValueMapper> keyValueMapper = (k, v) -> { - Message message = (Message) v; - BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); - String contentType = bindingProperties.getContentType(); - Map headers = new HashMap<>(((Message) v).getHeaders()); - if (!StringUtils.isEmpty(contentType)) { - headers.put(MessageHeaders.CONTENT_TYPE, contentType); - } - MessageHeaders messageHeaders = new MessageHeaders(headers); - return new KeyValue<>(k, - messageConverter.toMessage(message.getPayload(), - messageHeaders)); - }; - delegate = delegate.map(keyValueMapper); - } this.delegate = delegate; } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java index d2c34dcc0..5c6707f93 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamListenerParameterAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -16,16 +16,11 @@ package org.springframework.cloud.stream.binder.kstream; -import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KeyValueMapper; import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; import org.springframework.core.MethodParameter; import org.springframework.core.ResolvableType; -import org.springframework.messaging.Message; -import org.springframework.messaging.converter.MessageConverter; -import org.springframework.messaging.support.MessageBuilder; /** * @author Marius Bogoevici @@ -33,10 +28,10 @@ import org.springframework.messaging.support.MessageBuilder; */ public class KStreamListenerParameterAdapter implements StreamListenerParameterAdapter, KStream> { - private final MessageConverter messageConverter; + private final MessageConversionDelegate messageConversionDelegate; - public KStreamListenerParameterAdapter(MessageConverter messageConverter) { - this.messageConverter = messageConverter; + public KStreamListenerParameterAdapter(MessageConversionDelegate messageConversionDelegate) { + this.messageConversionDelegate = messageConversionDelegate; } @Override @@ -52,28 +47,7 @@ public class KStreamListenerParameterAdapter implements StreamListenerParameterA final Class valueClass = (resolvableType.getGeneric(1).getRawClass() != null) ? (resolvableType.getGeneric(1).getRawClass()) : Object.class; - return bindingTarget.map((KeyValueMapper) (o, o2) -> { - KeyValue keyValue; - if (valueClass.isAssignableFrom(o2.getClass())) { - keyValue = new KeyValue<>(o, o2); - } - else if (o2 instanceof Message) { - if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) { - keyValue = new KeyValue<>(o, ((Message) o2).getPayload()); - } - else { - keyValue = new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass)); - } - } - else if(o2 instanceof String || o2 instanceof byte[]) { - Message message = MessageBuilder.withPayload(o2).build(); - keyValue = new KeyValue<>(o, messageConverter.fromMessage(message, valueClass)); - } - else { - keyValue = new KeyValue<>(o, o2); - } - return keyValue; - }); + return bindingTarget.map(messageConversionDelegate.inboundKeyValueMapper(valueClass)); } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java index 843fd1b98..86e30e946 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamStreamListenerResultAdapter.java @@ -23,8 +23,6 @@ import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; -import org.springframework.messaging.Message; -import org.springframework.messaging.support.MessageBuilder; /** * @author Marius Bogoevici @@ -39,16 +37,7 @@ public class KStreamStreamListenerResultAdapter implements StreamListenerResultA @Override @SuppressWarnings("unchecked") public Closeable adapt(KStream streamListenerResult, KStreamBoundElementFactory.KStreamWrapper boundElement) { - boundElement.wrap(streamListenerResult.map((k, v) -> { - KeyValue keyValue; - if (v instanceof Message) { - keyValue = new KeyValue<>(k, v); - } - else { - keyValue = new KeyValue<>(k, MessageBuilder.withPayload(v).build()); - } - return keyValue; - })); + boundElement.wrap(streamListenerResult.map(KeyValue::new)); return new NoOpCloseable(); } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java new file mode 100644 index 000000000..9eee5e069 --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/MessageConversionDelegate.java @@ -0,0 +1,96 @@ +/* + * Copyright 2018 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 + * + * http://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.kstream; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.KeyValueMapper; + +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.MimeType; +import org.springframework.util.StringUtils; + +/** + * @author Soby Chacko + */ +public class MessageConversionDelegate { + + private final BindingServiceProperties bindingServiceProperties; + private final CompositeMessageConverterFactory compositeMessageConverterFactory; + + public MessageConversionDelegate(BindingServiceProperties bindingServiceProperties, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + this.bindingServiceProperties = bindingServiceProperties; + this.compositeMessageConverterFactory = compositeMessageConverterFactory; + } + + public KeyValueMapper> outboundKeyValueMapper(String name) { + BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name); + String contentType = bindingProperties.getContentType(); + MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory + .getMessageConverterForType(MimeType.valueOf(contentType)) + : null; + + return (k, v) -> { + Message message = v instanceof Message ? (Message)v : + MessageBuilder.withPayload(v).build(); + Map headers = new HashMap<>(message.getHeaders()); + if (!StringUtils.isEmpty(contentType)) { + headers.put(MessageHeaders.CONTENT_TYPE, contentType); + } + MessageHeaders messageHeaders = new MessageHeaders(headers); + return new KeyValue<>(k, + messageConverter.toMessage(message.getPayload(), + messageHeaders).getPayload()); + }; + } + + @SuppressWarnings("unchecked") + public KeyValueMapper> inboundKeyValueMapper(Class valueClass) { + MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered(); + return (KeyValueMapper) (o, o2) -> { + KeyValue keyValue; + if (valueClass.isAssignableFrom(o2.getClass())) { + keyValue = new KeyValue<>(o, o2); + } + else if (o2 instanceof Message) { + if (valueClass.isAssignableFrom(((Message) o2).getPayload().getClass())) { + keyValue = new KeyValue<>(o, ((Message) o2).getPayload()); + } + else { + keyValue = new KeyValue<>(o, messageConverter.fromMessage((Message) o2, valueClass)); + } + } + else if(o2 instanceof String || o2 instanceof byte[]) { + Message message = MessageBuilder.withPayload(o2).build(); + keyValue = new KeyValue<>(o, messageConverter.fromMessage(message, valueClass)); + } + else { + keyValue = new KeyValue<>(o, o2); + } + return keyValue; + }; + } +} diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java index 0db7dd0bb..4db42d63c 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kstream.config; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.kstream.Predicate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; @@ -26,6 +27,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kstream.KStreamBinder; +import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -43,6 +45,9 @@ public class KStreamBinderConfiguration { @Autowired private KafkaProperties kafkaProperties; + @Autowired(required = false) + private Predicate[] predicates; + @Bean public KafkaTopicProvisioner provisioningProvider(KafkaBinderConfigurationProperties binderConfigurationProperties) { return new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties); @@ -51,9 +56,14 @@ public class KStreamBinderConfiguration { @Bean public KStreamBinder kStreamBinder(KafkaBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, - KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig) { - return new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties, - streamsConfig); + KStreamExtendedBindingProperties kStreamExtendedBindingProperties, StreamsConfig streamsConfig, + MessageConversionDelegate messageConversionDelegate) { + KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, kStreamExtendedBindingProperties, + streamsConfig, messageConversionDelegate); + if (predicates != null) { + kStreamBinder.setPredicates(predicates); + } + return kStreamBinder; } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java index 4eae8c6b3..beb545c9f 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamBinderSupportAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -30,6 +30,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi import org.springframework.cloud.stream.binder.kstream.KStreamBoundElementFactory; import org.springframework.cloud.stream.binder.kstream.KStreamListenerParameterAdapter; import org.springframework.cloud.stream.binder.kstream.KStreamStreamListenerResultAdapter; +import org.springframework.cloud.stream.binder.kstream.MessageConversionDelegate; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.annotation.Bean; @@ -39,6 +40,7 @@ import org.springframework.util.ObjectUtils; /** * @author Marius Bogoevici + * @author Soby Chacko */ public class KStreamBinderSupportAutoConfiguration { @@ -56,8 +58,7 @@ public class KStreamBinderSupportAutoConfiguration { StreamsBuilderFactoryBean kStreamBuilderFactoryBean = new StreamsBuilderFactoryBean(streamsConfig); kStreamBuilderFactoryBean.setPhase(Integer.MAX_VALUE - 500); return kStreamBuilderFactoryBean; - } - else { + } else { throw new UnsatisfiedDependencyException(KafkaStreamsDefaultConfiguration.class.getName(), KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_BUILDER_BEAN_NAME, "streamsConfig", "There is no '" + KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME @@ -85,17 +86,20 @@ public class KStreamBinderSupportAutoConfiguration { @Bean public KStreamListenerParameterAdapter kafkaStreamListenerParameterAdapter( - CompositeMessageConverterFactory compositeMessageConverterFactory) { - return new KStreamListenerParameterAdapter( - compositeMessageConverterFactory.getMessageConverterForAllRegistered()); + MessageConversionDelegate messageConversionDelegate) { + return new KStreamListenerParameterAdapter(messageConversionDelegate); + } + + @Bean + public MessageConversionDelegate messageConversionDelegate(BindingServiceProperties bindingServiceProperties, + CompositeMessageConverterFactory compositeMessageConverterFactory) { + return new MessageConversionDelegate(bindingServiceProperties, compositeMessageConverterFactory); } @Bean public KStreamBoundElementFactory kafkaStreamBindableTargetFactory(StreamsBuilder kStreamBuilder, - BindingServiceProperties bindingServiceProperties, - CompositeMessageConverterFactory compositeMessageConverterFactory) { - return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties, - compositeMessageConverterFactory); + BindingServiceProperties bindingServiceProperties) { + return new KStreamBoundElementFactory(kStreamBuilder, bindingServiceProperties); } } diff --git a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java index a40e2c217..051dc43ad 100644 --- a/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java +++ b/spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/config/KStreamProducerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -18,7 +18,17 @@ package org.springframework.cloud.stream.binder.kstream.config; /** * @author Marius Bogoevici + * @author Soby Chacko */ public class KStreamProducerProperties extends KStreamCommonProperties { + private String additionalBranches; + + public String getAdditionalBranches() { + return additionalBranches; + } + + public void setAdditionalBranches(String additionalBranches) { + this.additionalBranches = additionalBranches; + } } diff --git a/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java new file mode 100644 index 000000000..a33a6752a --- /dev/null +++ b/spring-cloud-stream-binder-kstream/src/test/java/org/springframework/cloud/stream/binder/kstream/WordCountMultipleBranchesIntegrationTests.java @@ -0,0 +1,208 @@ +/* + * Copyright 2017 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 + * + * http://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.kstream; + +import java.util.Arrays; +import java.util.Date; +import java.util.Map; + +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.KeyValue; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.Predicate; +import org.apache.kafka.streams.kstream.TimeWindows; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +import org.junit.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binder.kstream.annotations.KStreamProcessor; +import org.springframework.cloud.stream.binder.kstream.config.KStreamApplicationSupportProperties; +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.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.messaging.handler.annotation.SendTo; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Marius Bogoevici + * @author Soby Chacko + * @author Gary Russell + */ +public class WordCountMultipleBranchesIntegrationTests { + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, "counts","foo","bar"); + + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { + Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts", "foo", "bar"); + } + + @AfterClass + public static void tearDown() { + consumer.close(); + } + + @Test + public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebEnvironment(false); + + ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.cloud.stream.bindings.input.destination=words", + "--spring.cloud.stream.bindings.output.destination=counts", + "--spring.cloud.stream.bindings.output.contentType=application/json", + "--spring.cloud.stream.kstream.binder.configuration.commit.interval.ms=1000", + "--spring.cloud.stream.kstream.binder.configuration.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.binder.configuration.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kstream.bindings.output.producer.additionalBranches=foo,bar", + "--spring.cloud.stream.bindings.output.producer.headerMode=raw", + "--spring.cloud.stream.bindings.input.consumer.headerMode=raw", + "--spring.cloud.stream.kstream.timeWindow.length=5000", + "--spring.cloud.stream.kstream.timeWindow.advanceBy=0", + "--spring.cloud.stream.kstream.binder.brokers=" + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kstream.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString()); + try { + receiveAndValidate(context); + } finally { + context.close(); + } + } + + private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("words"); + template.sendDefault("english"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, "counts"); + assertThat(cr.value().contains("\"word\":\"english\",\"count\":1")).isTrue(); + + template.sendDefault("french"); + template.sendDefault("french"); + cr = KafkaTestUtils.getSingleRecord(consumer, "foo"); + assertThat(cr.value().contains("\"word\":\"french\",\"count\":2")).isTrue(); + + template.sendDefault("spanish"); + template.sendDefault("spanish"); + template.sendDefault("spanish"); + cr = KafkaTestUtils.getSingleRecord(consumer, "bar"); + assertThat(cr.value().contains("\"word\":\"spanish\",\"count\":3")).isTrue(); + } + + @EnableBinding(KStreamProcessor.class) + @EnableAutoConfiguration + @EnableConfigurationProperties(KStreamApplicationSupportProperties.class) + public static class WordCountProcessorApplication { + + @Autowired + private TimeWindows timeWindows; + + @StreamListener("input") + @SendTo("output") + public KStream process(KStream input) { + return input + .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) + .groupBy((key, value) -> value) + .windowedBy(timeWindows) + .count(Materialized.as("WordCounts-multi")) + .toStream() + .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, new Date(key.window().start()), new Date(key.window().end())))); + } + + @Bean + public Predicate[] predicates() { + Predicate isEnglish = (k, v) -> v.word.equals("english"); + Predicate isFrench = (k, v) -> v.word.equals("french"); + Predicate isSpanish = (k, v) -> v.word.equals("spanish"); + return new Predicate[] {isEnglish, isFrench, isSpanish}; + } + + } + + static class WordCount { + + private String word; + + private long count; + + private Date start; + + private Date end; + + WordCount(String word, long count, Date start, Date end) { + this.word = word; + this.count = count; + this.start = start; + this.end = end; + } + + public String getWord() { + return word; + } + + public void setWord(String word) { + this.word = word; + } + + public long getCount() { + return count; + } + + public void setCount(long count) { + this.count = count; + } + + public Date getStart() { + return start; + } + + public void setStart(Date start) { + this.start = start; + } + + public Date getEnd() { + return end; + } + + public void setEnd(Date end) { + this.end = end; + } + } + +}