Support for KStream branch and split

Resolves #265 - Ensure that branching can be done and any message conversions done on all outbound data to multiple topics - Refine MessageConversion logic in KStream binder - Refactor message conversion code into a specific class - Cleanup and refactoring - Test to verify branching support - More refinements in the way nativeEncoding/nativeDecoding logic works in KStream binder - Doc changes for KStream binder
2018-01-05 18:54:10 -05:00
parent f2abb59b19
commit 86146bfe81
10 changed files with 600 additions and 143 deletions
--- 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
 </dependency>
 ----

-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<?, String> process(KStream<?, String> 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<?, WordCount> process(KStream<Object, String> 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<?, WordCount> isEnglish = (k, v) -> v.word.equals("english");
+        Predicate<?, WordCount> isFrench =  (k, v) -> v.word.equals("french");
+        Predicate<?, WordCount> 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

--- 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<KStream<Object, Object>> doBindConsumer(String name, String group,
-															KStream<Object, Object> inputTarget, ExtendedConsumerProperties<KStreamConsumerProperties> properties) {
+															KStream<Object, Object> inputTarget,
+															ExtendedConsumerProperties<KStreamConsumerProperties> properties) {

-		ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties = new ExtendedConsumerProperties<KafkaConsumerProperties>(
+		ExtendedConsumerProperties<KafkaConsumerProperties> 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<KStream<Object, Object>> doBindProducer(String name, KStream<Object, Object> outboundBindTarget,
 															ExtendedProducerProperties<KStreamProducerProperties> properties) {
-		ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<KafkaProducerProperties>(
+		ExtendedProducerProperties<KafkaProducerProperties> extendedProducerProperties = new ExtendedProducerProperties<>(
 				new KafkaProducerProperties());
 		this.kafkaTopicProvisioner.provisionProducerDestination(name, extendedProducerProperties);
-		outboundBindTarget = outboundBindTarget
-				.map((k, v) -> KeyValue.pair(k, ((Message<Object>) v).getPayload()));

-		Serde<?> keySerde = Serdes.ByteArray();
-		Serde<?> valueSerde = Serdes.ByteArray();
-		if (properties.isUseNativeEncoding()) {
-			outboundBindTarget.to(name, Produced.with((Serde<Object>) keySerde, (Serde<Object>) 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<Object>) keySerde, (Serde<Object>) valueSerde, branches);
+
+		return new DefaultBinding<>(name, null, outboundBindTarget, null);
+	}
+
+	private Serde<?> getKeySerde(ExtendedProducerProperties<KStreamProducerProperties> 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<KStreamProducerProperties> 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<Object>) keySerde, (Serde<Object>) 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<Object, Object> outboundBindTarget,
+				Serde<Object> keySerde, Serde<Object> valueSerde, String[] branches) {
+		KeyValueMapper<Object, Object, KeyValue<Object, Object>> 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;
+	}
+
 }
--- 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<KSt

 	private final BindingServiceProperties bindingServiceProperties;

-	private CompositeMessageConverterFactory compositeMessageConverterFactory;
-
-	public KStreamBoundElementFactory(StreamsBuilder streamBuilder, BindingServiceProperties bindingServiceProperties,
-									CompositeMessageConverterFactory compositeMessageConverterFactory) {
+	public KStreamBoundElementFactory(StreamsBuilder kStreamBuilder, BindingServiceProperties bindingServiceProperties) {
 		super(KStream.class);
 		this.bindingServiceProperties = bindingServiceProperties;
-		this.kStreamBuilder = streamBuilder;
-		this.compositeMessageConverterFactory = compositeMessageConverterFactory;
+		this.kStreamBuilder = kStreamBuilder;
 	}

 	@Override
@@ -67,7 +55,7 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KSt
 			KeyValue<Object, Object> keyValue;
 			BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
 			String contentType = bindingProperties.getContentType();
-			if (!StringUtils.isEmpty(contentType)) {
+			if (!StringUtils.isEmpty(contentType) && !bindingProperties.getConsumer().isUseNativeDecoding()) {
 				Message<Object> message = MessageBuilder.withPayload(value)
 						.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
 				keyValue = new KeyValue<>(key, message);
@@ -83,13 +71,9 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KSt
 	@Override
 	@SuppressWarnings("unchecked")
 	public KStream createOutput(final String name) {
-		BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
-		String contentType = bindingProperties.getContentType();
-		MessageConverter messageConverter = StringUtils.hasText(contentType) ? compositeMessageConverterFactory
-				.getMessageConverterForType(MimeType.valueOf(contentType)) : null;
-		KStreamWrapperHandler handler = new KStreamWrapperHandler(messageConverter, bindingServiceProperties, name);
+		KStreamWrapperHandler wrapper= new KStreamWrapperHandler();
 		ProxyFactory proxyFactory = new ProxyFactory(KStreamWrapper.class, KStream.class);
-		proxyFactory.addAdvice(handler);
+		proxyFactory.addAdvice(wrapper);
 		return (KStream) proxyFactory.getProxy();
 	}

@@ -102,39 +86,9 @@ public class KStreamBoundElementFactory extends AbstractBindingTargetFactory<KSt

 		private KStream<Object, Object> 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<Object, Object> 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<Object, Object, KeyValue<Object, Object>> keyValueMapper = (k, v) -> {
-					Message<?> message = (Message<?>) v;
-					BindingProperties bindingProperties = bindingServiceProperties.getBindingProperties(name);
-					String contentType = bindingProperties.getContentType();
-					Map<String, Object> 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;
 		}

--- 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<?,?>, 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<Object, Object> 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<Object> 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));
 	}

 }
--- 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<Object, Object> 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();
 	}

--- 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
@@ -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<Object, Object, KeyValue<Object, Object>> 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<String, Object> 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<Object, Object, KeyValue<Object, Object>> inboundKeyValueMapper(Class<?> valueClass) {
+		MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
+		return (KeyValueMapper) (o, o2) -> {
+			KeyValue<Object, Object> 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<Object> message = MessageBuilder.withPayload(o2).build();
+				keyValue =  new KeyValue<>(o, messageConverter.fromMessage(message, valueClass));
+			}
+			else {
+				keyValue =  new KeyValue<>(o, o2);
+			}
+			return keyValue;
+		};
+	}
+}
--- 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;
 	}

 }
--- 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);
 	}

 }
--- 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;
+	}
 }
--- 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
@@ -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<String, String> consumer;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		DefaultKafkaConsumerFactory<String, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
+		template.setDefaultTopic("words");
+		template.sendDefault("english");
+		ConsumerRecord<String, String> 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<?, WordCount> process(KStream<Object, String> 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<?, WordCount> isEnglish = (k, v) -> v.word.equals("english");
+			Predicate<?, WordCount> isFrench =  (k, v) -> v.word.equals("french");
+			Predicate<?, WordCount> 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;
+		}
+	}
+
+}