[[event-type-based-routing-in-kafka-streams-applications]]
= 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`.

When using the event routing feature in Kafkfa Streams binder, it uses the byte array `Serde` to deserialze all incoming records.
If the record headers match the event type, then only it uses the actual `Serde` to do a proper deserialization using either the configured or the inferred `Serde`.
This introduces issues if you set a deserialization exception handler on the binding as the expected deserialization only happens down the stack causing unexpected errors.
In order to address this issue, you can set the following property on the binding to force the binder to use the configured or inferred `Serde` instead of byte array `Serde`.

`spring.cloud.stream.kafka.streams.bindings.<process-in-0>.consumer.useConfiguredSerdeWhenRoutingEvents`

This way, the application can detect deserialization issues right away when using the event routing feature and can take appropriate handling decisions.

