From 0612ea85f7cd1973efe2a0e184c274bb111506cf Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 24 Jan 2023 21:49:45 -0500 Subject: [PATCH] Event type routing and deserialization issues (#2633) * Event type routing and deserialization issues In Kafka Streams binder, deserialization exception handler does not take effect when event type routing is enabled. Fixing this issue by allowing the applications to use the configured or inferred Serde rather than the byte[] Serde used by the event type router initially. Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2613 * Addressing PR review --- .../AbstractKafkaStreamsBinderProcessor.java | 28 +-- .../KafkaStreamsConsumerProperties.java | 16 +- ...ventTypeRoutingWithInferredSerdeTests.java | 165 ++++++++++++++++++ .../main/asciidoc/kafka/kafka-streams.adoc | 9 + 4 files changed, 207 insertions(+), 11 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java 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.