Event type based routing in Kafka Streams binder

Introducing the capability of routing records based on
event types. If a header in the incoming record contains
the event type set on the binding, then the function
associated with that binding gets invoked.

Adding test/docs.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003

docs/src/main/asciidoc/kafka-streams.adoc

@@ -1579,7 +1579,6 @@ By default, the `Kafkastreams.cleanup()` method is called when the binding is st
 See https://docs.spring.io/spring-kafka/reference/html/_reference.html#_configuration[the Spring Kafka documentation].
 To modify this behavior simply add a single `CleanupConfig` `@Bean` (configured to clean up on start, stop, or neither) to the application context; the bean will be detected and wired into the factory bean.
 
-
 === Kafka Streams topology visualization
 
 Kafka Streams binder provides the following actuator endpoints for retrieving the topology description using which you can visualize the topology using external tools.
@@ -1592,6 +1591,39 @@ You need to include the actuator and web dependencies from Spring Boot to access
 Further, you also need to add `kafkastreamstopology` to `management.endpoints.web.exposure.include` property.
 By default, the `kafkastreamstopology` endpoint is disabled.
 
+=== Event type based routing in Kafka Streams applications
+
+Routing functions available in regular message channel based binders are not supported in Kafka Streams binder.
+However, Kafka Streams binder still provides routing capabilities through the event type record header on the inbound records.
+
+To enable routing based on event types, the application must provide the following property.
+
+`spring.cloud.stream.kafka.streams.bindings.<binding-name>.consumer.eventTypes`.
+
+This can be a comma separated value.
+
+For example, lets assume we have this function:
+
+```
+@Bean
+public Function<KStream<Integer, Foo>, KStream<Integer, Foo>> process() {
+    return input -> input;
+}
+```
+
+Let us also assume that we only want the business logic in this function to be executed, if the incoming record has event types as `foo` or `bar`.
+That can be expressed as below using the `eventTypes` property on the binding.
+
+`spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypes=foo,bar`
+
+Now, when the application runs, the binder checks each incoming records for the header `event_type` and see if it has value set as `foo` or `bar`.
+If it does not find either of them, then the function execution will be skipped.
+
+By default, the binder expects the record header key to be `event_type`, but that can be changed per binding.
+For instance, if we want to change the header key on this binding to `my_event` instead of the default, that can be changed as below.
+
+`spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypeHeaderKey=my_event`.
+
 === Configuration Options
 
 This section contains the configuration options used by the Kafka Streams binder.
@@ -1603,9 +1635,9 @@ For common configuration options and properties pertaining to binder, refer to t
 The following properties are available at the binder level and must be prefixed with `spring.cloud.stream.kafka.streams.binder.`
 
 configuration::
-  Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
-  This property must be prefixed with `spring.cloud.stream.kafka.streams.binder.`.
- Following are some examples of using this property.
+Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
+This property must be prefixed with `spring.cloud.stream.kafka.streams.binder.`.
+Following are some examples of using this property.
 
 [source]
 ----
@@ -1621,56 +1653,56 @@ If you have more than processors in the application, all of them will acquire th
 In the case of properties like `application.id`, this will become problematic and therefore you have to carefully examine how the properties from `StreamsConfig` are mapped using this binder level `configuration` property.
 
 functions.<function-bean-name>.applicationId::
-  Applicable only for functional style processors.
-  This can be used for setting application ID per function in the application.
-  In the case of multiple functions, this is a handy way to set the application ID.
+Applicable only for functional style processors.
+This can be used for setting application ID per function in the application.
+In the case of multiple functions, this is a handy way to set the application ID.
 
 functions.<function-bean-name>.configuration::
-  Applicable only for functional style processors.
-  Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
-  This is similar to the binder level `configuration` property describe above, but this level of `configuration` property is restricted only against the named function.
-  When you have multiple processors and you want to restrict access to the configuration based on particular functions, you might want to use this.
-  All `StreamsConfig` properties can be used here.
+Applicable only for functional style processors.
+Map with a key/value pair containing properties pertaining to Apache Kafka Streams API.
+This is similar to the binder level `configuration` property describe above, but this level of `configuration` property is restricted only against the named function.
+When you have multiple processors and you want to restrict access to the configuration based on particular functions, you might want to use this.
+All `StreamsConfig` properties can be used here.
 
 brokers::
- Broker URL
+Broker URL
 +
 Default: `localhost`
 zkNodes::
- Zookeeper URL
+Zookeeper URL
 +
 Default: `localhost`
 
 deserializationExceptionHandler::
- Deserialization error handler type.
- This handler is applied at the binder level and thus applied against all input binding in the application.
- There is a way to control it in a more fine-grained way at the consumer binding level.
- Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
+Deserialization error handler type.
+This handler is applied at the binder level and thus applied against all input binding in the application.
+There is a way to control it in a more fine-grained way at the consumer binding level.
+Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
 +
 Default: `logAndFail`
 
 applicationId::
- Convenient way to set the application.id for the Kafka Streams application globally at the binder level.
- If the application contains multiple functions or `StreamListener` methods, then the application id should be set differently.
- See above where setting the application id is discussed in detail.
+Convenient way to set the application.id for the Kafka Streams application globally at the binder level.
+If the application contains multiple functions or `StreamListener` methods, then the application id should be set differently.
+See above where setting the application id is discussed in detail.
 +
 Default: application will generate a static application ID. See the application ID section for more details.
 
 stateStoreRetry.maxAttempts::
-  Max attempts for trying to connect to a state store.
+Max attempts for trying to connect to a state store.
 +
 Default: 1
 
 stateStoreRetry.backoffPeriod::
-  Backoff period when trying to connect to a state store on a retry.
+Backoff period when trying to connect to a state store on a retry.
 +
 Default: 1000 ms
 
 consumerProperties::
- Arbitrary consumer properties at the binder level.
+Arbitrary consumer properties at the binder level.
 
 producerProperties::
- Arbitrary producer properties at the binder level.
+Arbitrary producer properties at the binder level.
 
 ==== Kafka Streams Producer Properties
 
@@ -1678,23 +1710,23 @@ The following properties are _only_ available for Kafka Streams producers and mu
 For convenience, if there are multiple output bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.producer.`.
 
 keySerde::
-  key serde to use
+key serde to use
 +
 Default: See the above discussion on message de/serialization
 
 valueSerde::
-  value serde to use
+value serde to use
 +
 Default: See the above discussion on message de/serialization
 
 useNativeEncoding::
-  flag to enable/disable native encoding
+flag to enable/disable native encoding
 +
 Default: `true`.
 
 streamPartitionerBeanName:
- Custom outbound partitioner bean name to be used at the consumer.
- Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one.
+Custom outbound partitioner bean name to be used at the consumer.
+Applications can provide custom `StreamPartitioner` as a Spring bean and the name of this bean can be provided to the producer to use instead of the default one.
 +
 Default: See the discussion above on outbound partition support.
 
@@ -1704,40 +1736,40 @@ The following properties are available for Kafka Streams consumers and must be p
 For convenience, if there are multiple input bindings and they all require a common value, that can be configured by using the prefix `spring.cloud.stream.kafka.streams.default.consumer.`.
 
 applicationId::
- Setting application.id per input binding. This is only preferred for `StreamListener` based processors, for function based processors see other approaches outlined above.
+Setting application.id per input binding. This is only preferred for `StreamListener` based processors, for function based processors see other approaches outlined above.
 +
 Default: See above.
 
 keySerde::
-  key serde to use
+key serde to use
 +
 Default: See the above discussion on message de/serialization
 
 valueSerde::
-  value serde to use
+value serde to use
 +
 Default: See the above discussion on message de/serialization
 
 materializedAs::
-  state store to materialize when using incoming KTable types
+state store to materialize when using incoming KTable types
 +
 Default: `none`.
 
 useNativeDecoding::
-  flag to enable/disable native decoding
+flag to enable/disable native decoding
 +
 Default: `true`.
 
 dlqName::
-  DLQ topic name.
+DLQ topic name.
 +
 Default: See above on the discussion of error handling and DLQ.
 
 startOffset::
-  Offset to start from if there is no committed offset to consume from.
-  This is mostly used when the consumer is consuming from a topic for the first time.
-  Kafka Streams uses `earliest` as the default strategy and the binder uses the same default.
-  This can be overridden to `latest` using this property.
+Offset to start from if there is no committed offset to consume from.
+This is mostly used when the consumer is consuming from a topic for the first time.
+Kafka Streams uses `earliest` as the default strategy and the binder uses the same default.
+This can be overridden to `latest` using this property.
 +
 Default: `earliest`.
 
@@ -1745,18 +1777,28 @@ Note: Using `resetOffsets` on the consumer does not have any effect on Kafka Str
 Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand.
 
 deserializationExceptionHandler::
- Deserialization error handler type.
- This handler is applied per consumer binding as opposed to the binder level property described before.
- Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
+Deserialization error handler type.
+This handler is applied per consumer binding as opposed to the binder level property described before.
+Possible values are - `logAndContinue`, `logAndFail` or `sendToDlq`
 +
 Default: `logAndFail`
 
 timestampExtractorBeanName::
- Specific time stamp extractor bean name to be used at the consumer.
- Applications can provide `TimestampExtractor` as a Spring bean and the name of this bean can be provided to the consumer to use instead of the default one.
+Specific time stamp extractor bean name to be used at the consumer.
+Applications can provide `TimestampExtractor` as a Spring bean and the name of this bean can be provided to the consumer to use instead of the default one.
 +
 Default: See the discussion above on timestamp extractors.
 
+eventTypes::
+Comma separated list of supported event types for this binding.
++
+Default: `none`
+
+eventTypeHeaderKey::
+Event type header key on each incoming records through this binding.
++
+Default: `event_type`
+
 ==== Special note on concurrency
 
 In Kafka Streams, you can control of the number of threads a processor can create using the `num.stream.threads` property.

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java

@@ -19,11 +19,14 @@ package org.springframework.cloud.stream.binder.kafka.streams;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.common.utils.Bytes;
@@ -37,6 +40,8 @@ import org.apache.kafka.streams.kstream.GlobalKTable;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.Materialized;
+import org.apache.kafka.streams.processor.Processor;
+import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 import org.apache.kafka.streams.state.KeyValueStore;
 import org.apache.kafka.streams.state.StoreBuilder;
@@ -396,25 +401,13 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
 		}
 	}
 
+	@SuppressWarnings({"rawtypes", "unchecked"})
 	protected KStream<?, ?> getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties,
-			StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
+									StreamsBuilder streamsBuilder, Serde<?> keySerde, Serde<?> valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) {
 		if (firstBuild) {
 			addStateStoreBeans(streamsBuilder);
 		}
 
-		KStream<?, ?> stream;
-		if (this.kafkaStreamsExtendedBindingProperties
-				.getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) {
-			final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName));
-			stream = streamsBuilder.stream(pattern);
-		}
-		else {
-			String[] bindingTargets = StringUtils.commaDelimitedListToStringArray(
-					this.bindingServiceProperties.getBindingDestination(inboundName));
-			final Consumed<?, ?> consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerde, autoOffsetReset);
-			stream = streamsBuilder.stream(Arrays.asList(bindingTargets),
-					consumed);
-		}
 		final boolean nativeDecoding = this.bindingServiceProperties
 				.getConsumerProperties(inboundName).isUseNativeDecoding();
 		if (nativeDecoding) {
@@ -426,6 +419,62 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application
 					+ ". Inbound message conversion done by Spring Cloud Stream.");
 		}
 
+		KStream<?, ?> stream;
+		if (this.kafkaStreamsExtendedBindingProperties
+				.getExtendedConsumerProperties(inboundName).isDestinationIsPattern()) {
+			final Pattern pattern = Pattern.compile(this.bindingServiceProperties.getBindingDestination(inboundName));
+			stream = streamsBuilder.stream(pattern);
+		}
+		else {
+			String[] bindingTargets = StringUtils.commaDelimitedListToStringArray(
+					this.bindingServiceProperties.getBindingDestination(inboundName));
+			final Serde<?> valueSerdeToUse = StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes()) ?
+					new Serdes.BytesSerde() : valueSerde;
+			final Consumed<?, ?> consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerdeToUse, autoOffsetReset);
+			stream = streamsBuilder.stream(Arrays.asList(bindingTargets),
+					consumed);
+		}
+		//Check to see if event type based routing is enabled.
+		//See this issue for more context: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003
+		if (StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes())) {
+			AtomicBoolean matched = new AtomicBoolean();
+			// Processor to retrieve the header value.
+			stream.process(() -> new Processor() {
+
+				ProcessorContext context;
+
+				@Override
+				public void init(ProcessorContext context) {
+					this.context = context;
+				}
+
+				@Override
+				public void process(Object key, Object value) {
+					final Headers headers = this.context.headers();
+					final Iterable<Header> eventTypeHeader = headers.headers(kafkaStreamsConsumerProperties.getEventTypeHeaderKey());
+					if (eventTypeHeader != null && eventTypeHeader.iterator().hasNext()) {
+						String eventTypeFromHeader = new String(eventTypeHeader.iterator().next().value());
+						final String[] eventTypesFromBinding = StringUtils.commaDelimitedListToStringArray(kafkaStreamsConsumerProperties.getEventTypes());
+						for (String eventTypeFromBinding : eventTypesFromBinding) {
+							if (eventTypeFromHeader.equals(eventTypeFromBinding)) {
+								matched.set(true);
+								break;
+							}
+						}
+					}
+				}
+
+				@Override
+				public void close() {
+
+				}
+			});
+			// Branching based on event type match.
+			final KStream<?, ?>[] branch = stream.branch((key, value) -> matched.getAndSet(false));
+			// Deserialize if we have a branch from above.
+			final KStream<?, Object> deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize(null, ((Bytes) value).get()));
+			return getkStream(bindingProperties, deserializedKStream, nativeDecoding);
+		}
 		return getkStream(bindingProperties, stream, nativeDecoding);
 	}
 

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java

@@ -54,6 +54,18 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
 	 */
 	private String timestampExtractorBeanName;
 
+	/**
+	 * Comma separated list of supported event types for this binding.
+	 */
+	private String eventTypes;
+
+	/**
+	 * Record level header key for event type.
+	 * If the default value is overridden, then that is expected on each record header if eventType based
+	 * routing is enabled on this binding (by setting eventTypes).
+	 */
+	private String eventTypeHeaderKey = "event_type";
+
 	public String getApplicationId() {
 		return this.applicationId;
 	}
@@ -101,4 +113,20 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties {
 	public void setDeserializationExceptionHandler(DeserializationExceptionHandler deserializationExceptionHandler) {
 		this.deserializationExceptionHandler = deserializationExceptionHandler;
 	}
+
+	public String getEventTypes() {
+		return eventTypes;
+	}
+
+	public void setEventTypes(String eventTypes) {
+		this.eventTypes = eventTypes;
+	}
+
+	public String getEventTypeHeaderKey() {
+		return this.eventTypeHeaderKey;
+	}
+
+	public void setEventTypeHeaderKey(String eventTypeHeaderKey) {
+		this.eventTypeHeaderKey = eventTypeHeaderKey;
+	}
 }

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java

@@ -0,0 +1,190 @@
+/*
+ * Copyright 2019-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka.streams;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.streams.kstream.KStream;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.serializer.JsonDeserializer;
+import org.springframework.kafka.support.serializer.JsonSerializer;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class KafkaStreamsEventTypeRoutingTests {
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true,
+			"foo-1", "foo-2");
+
+	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+
+	private static Consumer<Integer, Foo> consumer;
+
+	@BeforeClass
+	public static void setUp() {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false",
+				embeddedKafka);
+		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+		consumerProps.put("value.deserializer", JsonDeserializer.class);
+		consumerProps.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
+		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
+		consumer = cf.createConsumer();
+		embeddedKafka.consumeFromEmbeddedTopics(consumer, "foo-2");
+	}
+
+	@AfterClass
+	public static void tearDown() {
+		consumer.close();
+	}
+
+	//See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003 for more context on this test.
+	@Test
+	public void testRoutingWorksBasedOnEventTypes() {
+		SpringApplication app = new SpringApplication(EventTypeRoutingTestConfig.class);
+		app.setWebApplicationType(WebApplicationType.NONE);
+
+		try (ConfigurableApplicationContext context = app.run(
+				"--server.port=0",
+				"--spring.jmx.enabled=false",
+				"--spring.cloud.stream.function.definition=process",
+				"--spring.cloud.stream.bindings.process-in-0.destination=foo-1",
+				"--spring.cloud.stream.bindings.process-out-0.destination=foo-2",
+				"--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.eventTypes=foo,bar",
+				"--spring.cloud.stream.kafka.streams.binder.functions.process.applicationId=process-id-foo-0",
+				"--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000",
+				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) {
+			Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+			senderProps.put("value.serializer", JsonSerializer.class);
+			DefaultKafkaProducerFactory<Integer, Foo> pf = new DefaultKafkaProducerFactory<>(senderProps);
+			try {
+				KafkaTemplate<Integer, Foo> template = new KafkaTemplate<>(pf, true);
+				template.setDefaultTopic("foo-1");
+				Foo foo1 = new Foo();
+				foo1.setFoo("foo-1");
+				Headers headers = new RecordHeaders();
+				headers.add(new RecordHeader("event_type", "foo".getBytes()));
+
+				final ProducerRecord<Integer, Foo> producerRecord1 = new ProducerRecord<>("foo-1", 0, 56, foo1, headers);
+				template.send(producerRecord1);
+
+				Foo foo2 = new Foo();
+				foo2.setFoo("foo-2");
+
+				final ProducerRecord<Integer, Foo> producerRecord2 = new ProducerRecord<>("foo-1", 0, 57, foo2);
+				template.send(producerRecord2);
+
+				Foo foo3 = new Foo();
+				foo3.setFoo("foo-3");
+
+				final ProducerRecord<Integer, Foo> producerRecord3 = new ProducerRecord<>("foo-1", 0, 58, foo3, headers);
+				template.send(producerRecord3);
+
+				Foo foo4 = new Foo();
+				foo4.setFoo("foo-4");
+				Headers headers1 = new RecordHeaders();
+				headers1.add(new RecordHeader("event_type", "bar".getBytes()));
+
+				final ProducerRecord<Integer, Foo> producerRecord4 = new ProducerRecord<>("foo-1", 0, 59, foo4, headers1);
+				template.send(producerRecord4);
+
+				final ConsumerRecords<Integer, Foo> records = KafkaTestUtils.getRecords(consumer);
+
+				assertThat(records.count()).isEqualTo(3);
+
+				List<Integer> keys = new ArrayList<>();
+				List<Foo> values = new ArrayList<>();
+
+				records.forEach(integerFooConsumerRecord -> {
+					keys.add(integerFooConsumerRecord.key());
+					values.add(integerFooConsumerRecord.value());
+				});
+
+				assertThat(keys).containsExactlyInAnyOrder(56, 58, 59);
+				assertThat(values).containsExactlyInAnyOrder(foo1, foo3, foo4);
+			}
+			finally {
+				pf.destroy();
+			}
+		}
+	}
+
+	@EnableAutoConfiguration
+	public static class EventTypeRoutingTestConfig {
+
+		@Bean
+		public Function<KStream<Integer, Foo>, KStream<Integer, Foo>> process() {
+			return input -> input;
+		}
+
+	}
+
+	static class Foo {
+		String foo;
+
+		public String getFoo() {
+			return foo;
+		}
+
+		public void setFoo(String foo) {
+			this.foo = foo;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+			Foo foo1 = (Foo) o;
+			return Objects.equals(foo, foo1.foo);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(foo);
+		}
+	}
+}