diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index 5facdb478..ba97e4057 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2022 the original author or authors. + * Copyright 2019-2023 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. @@ -409,7 +409,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } } - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({"unchecked"}) protected KStream getKStream(String inboundName, BindingProperties bindingProperties, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset, boolean firstBuild) { if (firstBuild) { @@ -428,8 +428,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application } KStream stream; - final Serde valueSerdeToUse = StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes()) ? - new Serdes.BytesSerde() : valueSerde; + final Serde valueSerdeToUse = getValueSerdeToUse(kafkaStreamsConsumerProperties, valueSerde); final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerdeToUse, autoOffsetReset); if (this.kafkaStreamsExtendedBindingProperties @@ -455,13 +454,21 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application // 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 deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize( - topicObject.get(), headersObject.get(), ((Bytes) value).get())); + final KStream deserializedKStream = !kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ? + branch[0].mapValues(value -> valueSerde.deserializer().deserialize( + topicObject.get(), headersObject.get(), ((Bytes) value).get())) : (KStream) branch[0]; return getkStream(bindingProperties, deserializedKStream, nativeDecoding); } return getkStream(bindingProperties, stream, nativeDecoding); } + private Serde getValueSerdeToUse(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, Serde valueSerde) { + if (StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes())) { + return kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ? valueSerde : new Serdes.BytesSerde(); + } + return valueSerde; + } + private KStream getkStream(BindingProperties bindingProperties, KStream stream, boolean nativeDecoding) { if (!nativeDecoding) { AtomicReference headersAtomicReference = new AtomicReference<>(); @@ -556,13 +563,13 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application consumed); } + @SuppressWarnings({"unchecked"}) private KTable getKTable(KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, String materializedAs, String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) { - final Serde valueSerdeToUse = StringUtils.hasText(kafkaStreamsConsumerProperties.getEventTypes()) ? - new Serdes.BytesSerde() : valueSerde; + final Serde valueSerdeToUse = getValueSerdeToUse(kafkaStreamsConsumerProperties, valueSerde); final Consumed consumed = getConsumed(kafkaStreamsConsumerProperties, keySerde, valueSerdeToUse, autoOffsetReset); @@ -586,8 +593,9 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application .noDefaultBranch(); final KStream[] branch = stringKStreamMap.values().toArray(new KStream[0]); // Deserialize if we have a branch from above. - final KStream deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize( - topicObject.get(), headersObject.get(), ((Bytes) value).get())); + final KStream deserializedKStream = !kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ? + branch[0].mapValues(value -> valueSerde.deserializer().deserialize( + topicObject.get(), headersObject.get(), ((Bytes) value).get())) : (KStream) branch[0]; return deserializedKStream.toTable(); } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java index 17b10a962..616fb207f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsConsumerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 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. @@ -71,6 +71,12 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { */ private String consumedAs; + /** + * When event type based routing is enabled, the binder uses the byte[] Serde by default. + * Use this property to override this default behavior by forcing the binder to use the configured or inferred Serde. + */ + private boolean useConfiguredSerdeWhenRoutingEvents; + public String getApplicationId() { return this.applicationId; } @@ -142,4 +148,12 @@ public class KafkaStreamsConsumerProperties extends KafkaConsumerProperties { public void setConsumedAs(String consumedAs) { this.consumedAs = consumedAs; } + + public boolean isUseConfiguredSerdeWhenRoutingEvents() { + return this.useConfiguredSerdeWhenRoutingEvents; + } + + public void setUseConfiguredSerdeWhenRoutingEvents(boolean useConfiguredSerdeWhenRoutingEvents) { + this.useConfiguredSerdeWhenRoutingEvents = useConfiguredSerdeWhenRoutingEvents; + } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java new file mode 100644 index 000000000..003e91de1 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java @@ -0,0 +1,165 @@ +/* + * Copyright 2023-2023 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.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.common.serialization.IntegerDeserializer; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.kstream.KStream; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * For more context around this test, see https://github.com/spring-cloud/spring-cloud-stream/issues/2613 + * + * @author Soby Chacko + */ +@EmbeddedKafka(topics = "foo-2") +public class EventTypeRoutingWithInferredSerdeTests { + + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + + private static Consumer consumer; + + @BeforeAll + public static void setUp() { + Map consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false", + embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put("value.deserializer", IntegerDeserializer.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "foo-2"); + } + + @AfterAll + public static void tearDown() { + consumer.close(); + } + + @Test + void eventTypeRoutingWorksWhenInferredSerdeThrowsException() { + SpringApplication app = new SpringApplication(EventTypeRoutingWithConfiguredSerdeConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + + try (ConfigurableApplicationContext context = app.run( + "--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.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.bindings.process-in-0.consumer.useConfiguredSerdeWhenRoutingEvents=true", + "--spring.cloud.stream.kafka.streams.binder.deserializationExceptionHandler=logAndContinue", + "--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 senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put("value.serializer", IntegerSerializer.class); + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("foo-1"); + + Headers headers = new RecordHeaders(); + headers.add(new RecordHeader("event_type", "foo".getBytes())); + + final ProducerRecord producerRecord1 = new ProducerRecord<>("foo-1", 0, 56, 56, headers); + template.send(producerRecord1); + + + final ProducerRecord producerRecord2 = new ProducerRecord<>("foo-1", 0, 57, 57); + template.send(producerRecord2); + + final ProducerRecord producerRecord3 = new ProducerRecord<>("foo-1", 0, 58, 58, headers); + template.send(producerRecord3); + + Headers headers1 = new RecordHeaders(); + headers1.add(new RecordHeader("event_type", "bar".getBytes())); + + final ProducerRecord producerRecord4 = new ProducerRecord<>("foo-1", 0, 59, 59, headers1); + template.send(producerRecord4); + + senderProps.put("value.serializer", StringSerializer.class); + DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); + template1.setDefaultTopic("foo-1"); + final ProducerRecord producerRecordBar = new ProducerRecord<>("foo-1", 0, 67, "foobar", headers); + // This send should throw a deserializer exception by the Kafka Streams processor. + // Since the processor is provisioned with an exception handler of logAndContinue, it shouldn't affect anything. + template1.send(producerRecordBar); + + final ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + + assertThat(records.count()).isEqualTo(3); + + List keys = new ArrayList<>(); + List values = new ArrayList<>(); + + records.forEach(integerFooConsumerRecord -> { + keys.add(integerFooConsumerRecord.key()); + values.add(integerFooConsumerRecord.value()); + }); + + assertThat(keys).containsExactlyInAnyOrder(56, 58, 59); + assertThat(values).containsExactlyInAnyOrder(56, 58, 59); + } + finally { + pf.destroy(); + } + } + } + + @EnableAutoConfiguration + public static class EventTypeRoutingWithConfiguredSerdeConfig { + + @Bean + public Function, KStream> process() { + return input -> input; + } + + } + +} diff --git a/docs/src/main/asciidoc/kafka/kafka-streams.adoc b/docs/src/main/asciidoc/kafka/kafka-streams.adoc index 884f0ce04..b4c6a9cb6 100644 --- a/docs/src/main/asciidoc/kafka/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka/kafka-streams.adoc @@ -1552,6 +1552,15 @@ For instance, if we want to change the header key on this binding to `my_event` `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..consumer.useConfiguredSerdeWhenRoutingEvents` + +This way, the application can detect deserialization issues right away when using the event routing feature and can take appropriate handling decisions. + === Binding visualization and control in Kafka Streams binder Starting with version 3.1.2, Kafka Streams binder supports binding visualization and control.