SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Event type routing and deserialization issues (#2633)

Browse Source 
* 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
This commit is contained in:
Soby Chacko
2023-01-24 21:49:45 -05:00
committed by GitHub
parent 64aef347e4
commit 0612ea85f7
4 changed files with 207 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java
View File

@@ -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<?, Object> deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
topicObject.get(), headersObject.get(), ((Bytes) value).get()));
final KStream<?, Object> deserializedKStream = !kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ?
branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
topicObject.get(), headersObject.get(), ((Bytes) value).get())) : (KStream<?, Object>) 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<Headers> 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<?, Object> deserializedKStream = branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
topicObject.get(), headersObject.get(), ((Bytes) value).get()));
final KStream<?, Object> deserializedKStream = !kafkaStreamsConsumerProperties.isUseConfiguredSerdeWhenRoutingEvents() ?
branch[0].mapValues(value -> valueSerde.deserializer().deserialize(
topicObject.get(), headersObject.get(), ((Bytes) value).get())) : (KStream<?, Object>) branch[0];
return deserializedKStream.toTable();
}

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

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

165
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java Normal file
View File

@@ -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<Integer, Integer> consumer;
@BeforeAll
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", IntegerDeserializer.class);
DefaultKafkaConsumerFactory<Integer, Integer> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
senderProps.put("value.serializer", IntegerSerializer.class);
DefaultKafkaProducerFactory<Integer, Integer> pf = new DefaultKafkaProducerFactory<>(senderProps);
try {
KafkaTemplate<Integer, Integer> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic("foo-1");
Headers headers = new RecordHeaders();
headers.add(new RecordHeader("event_type", "foo".getBytes()));
final ProducerRecord<Integer, Integer> producerRecord1 = new ProducerRecord<>("foo-1", 0, 56, 56, headers);
template.send(producerRecord1);
final ProducerRecord<Integer, Integer> producerRecord2 = new ProducerRecord<>("foo-1", 0, 57, 57);
template.send(producerRecord2);
final ProducerRecord<Integer, Integer> 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<Integer, Integer> producerRecord4 = new ProducerRecord<>("foo-1", 0, 59, 59, headers1);
template.send(producerRecord4);
senderProps.put("value.serializer", StringSerializer.class);
DefaultKafkaProducerFactory<Integer, String> pf1 = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template1 = new KafkaTemplate<>(pf1, true);
template1.setDefaultTopic("foo-1");
final ProducerRecord<Integer, String> 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<Integer, Integer> records = KafkaTestUtils.getRecords(consumer);
assertThat(records.count()).isEqualTo(3);
List<Integer> keys = new ArrayList<>();
List<Integer> 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<Integer, Integer>, KStream<Integer, Integer>> process() {
return input -> input;
}
}
}

9
docs/src/main/asciidoc/kafka/kafka-streams.adoc
View File

@@ -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.<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.
=== Binding visualization and control in Kafka Streams binder
Starting with version 3.1.2, Kafka Streams binder supports binding visualization and control.