diff --git a/docs/src/main/asciidoc/kafka-streams.adoc b/docs/src/main/asciidoc/kafka-streams.adoc index a7fa184c1..81646132f 100644 --- a/docs/src/main/asciidoc/kafka-streams.adoc +++ b/docs/src/main/asciidoc/kafka-streams.adoc @@ -876,69 +876,19 @@ When the above property is set, all the deserialization error records are automa [source] ---- -spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: foo-dlq +spring.cloud.stream.kafka.streams.bindings.input.consumer.dlqName: custom-dlq ---- -If this is set, then the error records are sent to the topic `foo-dlq`. If this is not set, then it will create a DLQ +If this is set, then the error records are sent to the topic `custom-dlq`. If this is not set, then it will create a DLQ topic with the name `error..`. A couple of things to keep in mind when using the exception handling feature in Kafka Streams binder. * The property `spring.cloud.stream.kafka.streams.binder.serdeError` is applicable for the entire application. This implies -that if there are multiple `StreamListener` methods in the same application, this property is applied to all of them. +that if there are multiple functions or `StreamListener` methods in the same application, this property is applied to all of them. * The exception handling for deserialization works consistently with native deserialization and framework provided message conversion. -==== Handling Non-Deserialization Exceptions - -For general error handling in Kafka Streams binder, it is up to the end user applications to handle application level errors. -As a side effect of providing a DLQ for deserialization exception handlers, Kafka Streams binder provides a way to get -access to the DLQ sending bean directly from your application. -Once you get access to that bean, you can programmatically send any exception records from your application to the DLQ. - -It continues to remain hard to robust error handling using the high-level DSL; Kafka Streams doesn't natively support error -handling yet. - -However, when you use the low-level Processor API in your application, there are options to control this behavior. See -below. - -[source] ----- -@Autowired -private SendToDlqAndContinue dlqHandler; - -@StreamListener("input") -@SendTo("output") -public KStream process(KStream input) { - - input.process(() -> new Processor() { - ProcessorContext context; - - @Override - public void init(ProcessorContext context) { - this.context = context; - } - - @Override - public void process(Object o, Object o2) { - - try { - ..... - ..... - } - catch(Exception e) { - //explicitly provide the kafka topic corresponding to the input binding as the first argument. - //DLQ handler will correctly map to the dlq topic from the actual incoming destination. - dlqHandler.sendToDlq("topic-name", (byte[]) o1, (byte[]) o2, context.partition()); - } - } - - ..... - ..... - }); -} ----- - === State Store State store is created automatically by Kafka Streams when the DSL is used. diff --git a/pom.xml b/pom.xml index fd2fb5a87..ad0392bbe 100644 --- a/pom.xml +++ b/pom.xml @@ -12,8 +12,8 @@ 1.8 - 2.3.0.RC1 - 3.2.0.RC1 + 2.3.0.RELEASE + 3.2.0.RELEASE 2.3.0 1.0.0.BUILD-SNAPSHOT 3.0.0.BUILD-SNAPSHOT diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index bb2e81bf0..2c1bb1b5b 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.Arrays; import java.util.Map; -import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -210,18 +209,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application concurrency); } - Map kafkaStreamsDlqDispatchers = applicationContext - .getBean("kafkaStreamsDlqDispatchers", Map.class); - - KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties) { - @Override - public Properties asProperties() { - Properties properties = super.asProperties(); - properties.put(SendToDlqAndContinue.KAFKA_STREAMS_DLQ_DISPATCHERS, - kafkaStreamsDlqDispatchers); - return properties; - } - }; + KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties); StreamsBuilderFactoryBean streamsBuilder = this.cleanupConfig == null ? new StreamsBuilderFactoryBean(kafkaStreamsConfiguration) @@ -236,6 +224,7 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition( "stream-builder-" + beanNamePostPrefix, streamsBuilderBeanDefinition); + extendedConsumerProperties.setApplicationId((String) streamConfigGlobalProperties.get(StreamsConfig.APPLICATION_ID_CONFIG)); //Removing the application ID from global properties so that the next function won't re-use it and cause race conditions. streamConfigGlobalProperties.remove(StreamsConfig.APPLICATION_ID_CONFIG); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java index 71655bc21..ec99fe660 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinder.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.Map; - import org.apache.kafka.streams.kstream.GlobalKTable; import org.springframework.cloud.stream.binder.AbstractBinder; @@ -54,8 +52,6 @@ public class GlobalKTableBinder extends private final KafkaTopicProvisioner kafkaTopicProvisioner; - private final Map kafkaStreamsDlqDispatchers; - // @checkstyle:off private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); @@ -63,11 +59,9 @@ public class GlobalKTableBinder extends public GlobalKTableBinder( KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - KafkaTopicProvisioner kafkaTopicProvisioner, - Map kafkaStreamsDlqDispatchers) { + KafkaTopicProvisioner kafkaTopicProvisioner) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; - this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers; } @Override @@ -76,12 +70,11 @@ public class GlobalKTableBinder extends String group, GlobalKTable inputTarget, ExtendedConsumerProperties properties) { if (!StringUtils.hasText(group)) { - group = this.binderConfigurationProperties.getApplicationId(); + group = properties.getExtension().getApplicationId(); } KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties, - this.kafkaStreamsDlqDispatchers); + this.binderConfigurationProperties, properties); return new DefaultBinding<>(name, group, inputTarget, null); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java index e8b65664a..e87d79b08 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java @@ -56,9 +56,9 @@ public class GlobalKTableBinderConfiguration { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - @Qualifier("kafkaStreamsDlqDispatchers") Map kafkaStreamsDlqDispatchers) { + @Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties) { GlobalKTableBinder globalKTableBinder = new GlobalKTableBinder(binderConfigurationProperties, - kafkaTopicProvisioner, kafkaStreamsDlqDispatchers); + kafkaTopicProvisioner); globalKTableBinder.setKafkaStreamsExtendedBindingProperties( kafkaStreamsExtendedBindingProperties); return globalKTableBinder; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java index d7a1c42a7..ac233b10e 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinder.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.Map; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.common.serialization.Serde; @@ -75,20 +73,16 @@ class KStreamBinder extends private final KeyValueSerdeResolver keyValueSerdeResolver; - private final Map kafkaStreamsDlqDispatchers; - KStreamBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsMessageConversionDelegate kafkaStreamsMessageConversionDelegate, KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, - KeyValueSerdeResolver keyValueSerdeResolver, - Map kafkaStreamsDlqDispatchers) { + KeyValueSerdeResolver keyValueSerdeResolver) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; this.kafkaStreamsMessageConversionDelegate = kafkaStreamsMessageConversionDelegate; this.kafkaStreamsBindingInformationCatalogue = KafkaStreamsBindingInformationCatalogue; this.keyValueSerdeResolver = keyValueSerdeResolver; - this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers; } @Override @@ -102,12 +96,11 @@ class KStreamBinder extends this.kafkaStreamsBindingInformationCatalogue.registerConsumerProperties(delegate, properties.getExtension()); if (!StringUtils.hasText(group)) { - group = this.binderConfigurationProperties.getApplicationId(); + group = properties.getExtension().getApplicationId(); } KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties, - this.kafkaStreamsDlqDispatchers); + this.binderConfigurationProperties, properties); return new DefaultBinding<>(name, group, inputTarget, null); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java index 33a8759bb..f8191478f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java @@ -16,9 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.Map; - -import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration; @@ -60,12 +57,10 @@ public class KStreamBinderConfiguration { KafkaStreamsMessageConversionDelegate KafkaStreamsMessageConversionDelegate, KafkaStreamsBindingInformationCatalogue KafkaStreamsBindingInformationCatalogue, KeyValueSerdeResolver keyValueSerdeResolver, - KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - @Qualifier("kafkaStreamsDlqDispatchers") Map kafkaStreamsDlqDispatchers) { + KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties) { KStreamBinder kStreamBinder = new KStreamBinder(binderConfigurationProperties, kafkaTopicProvisioner, KafkaStreamsMessageConversionDelegate, - KafkaStreamsBindingInformationCatalogue, keyValueSerdeResolver, - kafkaStreamsDlqDispatchers); + KafkaStreamsBindingInformationCatalogue, keyValueSerdeResolver); kStreamBinder.setKafkaStreamsExtendedBindingProperties( kafkaStreamsExtendedBindingProperties); return kStreamBinder; diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java index 1823acd4a..cb899e096 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinder.java @@ -16,8 +16,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.util.Map; - import org.apache.kafka.streams.kstream.KTable; import org.springframework.cloud.stream.binder.AbstractBinder; @@ -55,19 +53,15 @@ class KTableBinder extends private final KafkaTopicProvisioner kafkaTopicProvisioner; - private Map kafkaStreamsDlqDispatchers; - // @checkstyle:off private KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties = new KafkaStreamsExtendedBindingProperties(); // @checkstyle:on KTableBinder(KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - KafkaTopicProvisioner kafkaTopicProvisioner, - Map kafkaStreamsDlqDispatchers) { + KafkaTopicProvisioner kafkaTopicProvisioner) { this.binderConfigurationProperties = binderConfigurationProperties; this.kafkaTopicProvisioner = kafkaTopicProvisioner; - this.kafkaStreamsDlqDispatchers = kafkaStreamsDlqDispatchers; } @Override @@ -78,12 +72,11 @@ class KTableBinder extends ExtendedConsumerProperties properties) { // @checkstyle:on if (!StringUtils.hasText(group)) { - group = this.binderConfigurationProperties.getApplicationId(); + group = properties.getExtension().getApplicationId(); } KafkaStreamsBinderUtils.prepareConsumerBinding(name, group, getApplicationContext(), this.kafkaTopicProvisioner, - this.binderConfigurationProperties, properties, - this.kafkaStreamsDlqDispatchers); + this.binderConfigurationProperties, properties); return new DefaultBinding<>(name, group, inputTarget, null); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index 12e7ef427..91aa9197f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java @@ -56,9 +56,9 @@ public class KTableBinderConfiguration { KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, - @Qualifier("kafkaStreamsDlqDispatchers") Map kafkaStreamsDlqDispatchers) { + @Qualifier("streamConfigGlobalProperties") Map streamConfigGlobalProperties) { KTableBinder kTableBinder = new KTableBinder(binderConfigurationProperties, - kafkaTopicProvisioner, kafkaStreamsDlqDispatchers); + kafkaTopicProvisioner); kTableBinder.setKafkaStreamsExtendedBindingProperties(kafkaStreamsExtendedBindingProperties); return kTableBinder; } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 4a96c91b1..1d4b874eb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -55,6 +55,7 @@ import org.springframework.core.env.Environment; import org.springframework.core.env.MapPropertySource; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.core.CleanupConfig; +import org.springframework.kafka.streams.RecoveringDeserializationExceptionHandler; import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -152,7 +153,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @Bean("streamConfigGlobalProperties") public Map streamConfigGlobalProperties( KafkaStreamsBinderConfigurationProperties configProperties, - KafkaStreamsConfiguration kafkaStreamsConfiguration, ConfigurableEnvironment environment) { + KafkaStreamsConfiguration kafkaStreamsConfiguration, ConfigurableEnvironment environment, + SendToDlqAndContinue sendToDlqAndContinue) { Properties properties = kafkaStreamsConfiguration.asProperties(); @@ -213,19 +215,20 @@ public class KafkaStreamsBinderSupportAutoConfiguration { .getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndContinue) { properties.put( StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, - LogAndContinueExceptionHandler.class.getName()); + LogAndContinueExceptionHandler.class); } else if (configProperties .getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.logAndFail) { properties.put( StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, - LogAndFailExceptionHandler.class.getName()); + LogAndFailExceptionHandler.class); } else if (configProperties .getSerdeError() == KafkaStreamsBinderConfigurationProperties.SerdeError.sendToDlq) { properties.put( StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, - SendToDlqAndContinue.class.getName()); + RecoveringDeserializationExceptionHandler.class); + properties.put(RecoveringDeserializationExceptionHandler.KSTREAM_DESERIALIZATION_RECOVERER, sendToDlqAndContinue); } if (!ObjectUtils.isEmpty(configProperties.getConfiguration())) { @@ -346,11 +349,6 @@ public class KafkaStreamsBinderSupportAutoConfiguration { return new StreamsBuilderFactoryManager(catalogue, kafkaStreamsRegistry); } - @Bean("kafkaStreamsDlqDispatchers") - public Map dlqDispatchers() { - return new HashMap<>(); - } - @Bean @Conditional(FunctionDetectorCondition.class) public KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor(BindingServiceProperties bindingServiceProperties, diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java index 4f7fcb883..e595be511 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderUtils.java @@ -16,17 +16,29 @@ package org.springframework.cloud.stream.binder.kafka.streams; +import java.util.HashMap; import java.util.Map; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.ByteArraySerializer; import org.apache.kafka.streams.kstream.KStream; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.context.ApplicationContext; import org.springframework.core.MethodParameter; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -43,8 +55,7 @@ final class KafkaStreamsBinderUtils { static void prepareConsumerBinding(String name, String group, ApplicationContext context, KafkaTopicProvisioner kafkaTopicProvisioner, KafkaStreamsBinderConfigurationProperties binderConfigurationProperties, - ExtendedConsumerProperties properties, - Map kafkaStreamsDlqDispatchers) { + ExtendedConsumerProperties properties) { ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties<>( properties.getExtension()); if (binderConfigurationProperties @@ -59,33 +70,73 @@ final class KafkaStreamsBinderUtils { } if (extendedConsumerProperties.getExtension().isEnableDlq()) { - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = !StringUtils + + ProducerFactory producerFactory = getProducerFactory( + new ExtendedProducerProperties<>( + extendedConsumerProperties.getExtension().getDlqProducerProperties()), + binderConfigurationProperties); + KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFactory); + + + DeadLetterPublishingRecoverer kafkaStreamsBinderDlqRecoverer = !StringUtils .isEmpty(extendedConsumerProperties.getExtension().getDlqName()) - ? new KafkaStreamsDlqDispatch( - extendedConsumerProperties.getExtension() - .getDlqName(), - binderConfigurationProperties, - extendedConsumerProperties.getExtension()) - : null; + ? new DeadLetterPublishingRecoverer(kafkaTemplate, (cr, e) -> new TopicPartition(extendedConsumerProperties.getExtension() + .getDlqName(), cr.partition())) + : null; for (String inputTopic : inputTopics) { if (StringUtils.isEmpty( extendedConsumerProperties.getExtension().getDlqName())) { - String dlqName = "error." + inputTopic + "." + group; - kafkaStreamsDlqDispatch = new KafkaStreamsDlqDispatch(dlqName, - binderConfigurationProperties, - extendedConsumerProperties.getExtension()); + kafkaStreamsBinderDlqRecoverer = new DeadLetterPublishingRecoverer(kafkaTemplate, (cr, e) -> new TopicPartition("error." + inputTopic + "." + group, cr.partition())); } SendToDlqAndContinue sendToDlqAndContinue = context .getBean(SendToDlqAndContinue.class); sendToDlqAndContinue.addKStreamDlqDispatch(inputTopic, - kafkaStreamsDlqDispatch); - - kafkaStreamsDlqDispatchers.put(inputTopic, kafkaStreamsDlqDispatch); + kafkaStreamsBinderDlqRecoverer); } } } + private static DefaultKafkaProducerFactory getProducerFactory( + ExtendedProducerProperties producerProperties, + KafkaBinderConfigurationProperties configurationProperties) { + Map props = new HashMap<>(); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.ACKS_CONFIG, configurationProperties.getRequiredAcks()); + Map mergedConfig = configurationProperties + .mergedProducerConfiguration(); + if (!ObjectUtils.isEmpty(mergedConfig)) { + props.putAll(mergedConfig); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, + configurationProperties.getKafkaConnectionString()); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { + props.put(ProducerConfig.BATCH_SIZE_CONFIG, + String.valueOf(producerProperties.getExtension().getBufferSize())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { + props.put(ProducerConfig.LINGER_MS_CONFIG, + String.valueOf(producerProperties.getExtension().getBatchTimeout())); + } + if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { + props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, + producerProperties.getExtension().getCompressionType().toString()); + } + if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { + props.putAll(producerProperties.getExtension().getConfiguration()); + } + // Always send as byte[] on dlq (the same byte[] that the consumer received) + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + ByteArraySerializer.class); + + return new DefaultKafkaProducerFactory<>(props); + } + + static boolean supportsKStream(MethodParameter methodParameter, Class targetBeanClass) { return KStream.class.isAssignableFrom(targetBeanClass) && KStream.class.isAssignableFrom(methodParameter.getParameterType()); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java deleted file mode 100644 index b80c39359..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java +++ /dev/null @@ -1,152 +0,0 @@ -/* - * Copyright 2018-2019 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.HashMap; -import java.util.Map; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.clients.producer.ProducerRecord; -import org.apache.kafka.common.serialization.ByteArraySerializer; - -import org.springframework.cloud.stream.binder.ExtendedProducerProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.core.ProducerFactory; -import org.springframework.kafka.support.SendResult; -import org.springframework.util.ObjectUtils; -import org.springframework.util.concurrent.ListenableFuture; -import org.springframework.util.concurrent.ListenableFutureCallback; - -/** - * Send records in error to a DLQ. - * - * @author Soby Chacko - * @author Rafal Zukowski - * @author Gary Russell - */ -class KafkaStreamsDlqDispatch { - - private final Log logger = LogFactory.getLog(getClass()); - - private final KafkaTemplate kafkaTemplate; - - private final String dlqName; - - KafkaStreamsDlqDispatch(String dlqName, - KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties, - KafkaConsumerProperties kafkaConsumerProperties) { - ProducerFactory producerFactory = getProducerFactory( - new ExtendedProducerProperties<>( - kafkaConsumerProperties.getDlqProducerProperties()), - kafkaBinderConfigurationProperties); - - this.kafkaTemplate = new KafkaTemplate<>(producerFactory); - this.dlqName = dlqName; - } - - @SuppressWarnings("unchecked") - public void sendToDlq(byte[] key, byte[] value, int partittion) { - ProducerRecord producerRecord = new ProducerRecord<>(this.dlqName, - partittion, key, value, null); - - StringBuilder sb = new StringBuilder().append(" a message with key='") - .append(toDisplayString(ObjectUtils.nullSafeToString(key))).append("'") - .append(" and payload='") - .append(toDisplayString(ObjectUtils.nullSafeToString(value))).append("'") - .append(" received from ").append(partittion); - ListenableFuture> sentDlq = null; - try { - sentDlq = this.kafkaTemplate.send(producerRecord); - sentDlq.addCallback( - new ListenableFutureCallback>() { - - @Override - public void onFailure(Throwable ex) { - KafkaStreamsDlqDispatch.this.logger - .error("Error sending to DLQ " + sb.toString(), ex); - } - - @Override - public void onSuccess(SendResult result) { - if (KafkaStreamsDlqDispatch.this.logger.isDebugEnabled()) { - KafkaStreamsDlqDispatch.this.logger - .debug("Sent to DLQ " + sb.toString()); - } - } - }); - } - catch (Exception ex) { - if (sentDlq == null) { - KafkaStreamsDlqDispatch.this.logger - .error("Error sending to DLQ " + sb.toString(), ex); - } - } - } - - private DefaultKafkaProducerFactory getProducerFactory( - ExtendedProducerProperties producerProperties, - KafkaBinderConfigurationProperties configurationProperties) { - Map props = new HashMap<>(); - props.put(ProducerConfig.RETRIES_CONFIG, 0); - props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); - props.put(ProducerConfig.ACKS_CONFIG, configurationProperties.getRequiredAcks()); - Map mergedConfig = configurationProperties - .mergedProducerConfiguration(); - if (!ObjectUtils.isEmpty(mergedConfig)) { - props.putAll(mergedConfig); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) { - props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, - configurationProperties.getKafkaConnectionString()); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) { - props.put(ProducerConfig.BATCH_SIZE_CONFIG, - String.valueOf(producerProperties.getExtension().getBufferSize())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) { - props.put(ProducerConfig.LINGER_MS_CONFIG, - String.valueOf(producerProperties.getExtension().getBatchTimeout())); - } - if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) { - props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, - producerProperties.getExtension().getCompressionType().toString()); - } - if (!ObjectUtils.isEmpty(producerProperties.getExtension().getConfiguration())) { - props.putAll(producerProperties.getExtension().getConfiguration()); - } - // Always send as byte[] on dlq (the same byte[] that the consumer received) - props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - ByteArraySerializer.class); - - return new DefaultKafkaProducerFactory<>(props); - } - - private String toDisplayString(String original) { - if (original.length() <= 50) { - return original; - } - return original.substring(0, 50) + "..."; - } - -} diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java index 4a4b3b1b2..1ba37ed6a 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java @@ -22,6 +22,7 @@ import java.util.Map; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeader; @@ -65,6 +66,8 @@ public class KafkaStreamsMessageConversionDelegate { private final KafkaStreamsBinderConfigurationProperties kstreamBinderConfigurationProperties; + Exception[] failedWithDeserException = new Exception[1]; + KafkaStreamsMessageConversionDelegate( CompositeMessageConverter compositeMessageConverter, SendToDlqAndContinue sendToDlqAndContinue, @@ -200,6 +203,7 @@ public class KafkaStreamsMessageConversionDelegate { "Deserialization has failed. This will be skipped from further processing.", e); // pass through + failedWithDeserException[0] = e; } return isValidRecord; }, @@ -207,7 +211,7 @@ public class KafkaStreamsMessageConversionDelegate { // in the first filter above. (k, v) -> true); // process errors from the second filter in the branch above. - processErrorFromDeserialization(bindingTarget, branch[1]); + processErrorFromDeserialization(bindingTarget, branch[1], failedWithDeserException); // first branch above is the branch where the messages are converted, let it go // through further processing. @@ -264,7 +268,7 @@ public class KafkaStreamsMessageConversionDelegate { @SuppressWarnings({ "unchecked", "rawtypes" }) private void processErrorFromDeserialization(KStream bindingTarget, - KStream branch) { + KStream branch, Exception[] exception) { branch.process(() -> new Processor() { ProcessorContext context; @@ -279,7 +283,6 @@ public class KafkaStreamsMessageConversionDelegate { if (o2 != null) { if (KafkaStreamsMessageConversionDelegate.this.kstreamBindingInformationCatalogue .isDlqEnabled(bindingTarget)) { - String destination = this.context.topic(); if (o2 instanceof Message) { Message message = (Message) o2; @@ -288,15 +291,17 @@ public class KafkaStreamsMessageConversionDelegate { Serializer keySerializer = keySerde.serializer(); byte[] keyBytes = keySerializer.serialize(null, o); + ConsumerRecord consumerRecord = new ConsumerRecord(this.context.topic(), this.context.partition(), this.context.offset(), + keyBytes, message.getPayload()); + KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue - .sendToDlq(destination, keyBytes, - (byte[]) message.getPayload(), - this.context.partition()); + .sendToDlq(consumerRecord, exception[0]); } else { + ConsumerRecord consumerRecord = new ConsumerRecord(this.context.topic(), this.context.partition(), this.context.offset(), + o, o2); KafkaStreamsMessageConversionDelegate.this.sendToDlqAndContinue - .sendToDlq(destination, (byte[]) o, (byte[]) o2, - this.context.partition()); + .sendToDlq(consumerRecord, exception[0]); } } else if (KafkaStreamsMessageConversionDelegate.this.kstreamBinderConfigurationProperties diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java index 4dc818ff9..569ef8c4f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/SendToDlqAndContinue.java @@ -16,109 +16,48 @@ package org.springframework.cloud.stream.binder.kafka.streams; -import java.lang.reflect.Field; import java.util.HashMap; import java.util.Map; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.consumer.OffsetAndMetadata; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.streams.errors.DeserializationExceptionHandler; -import org.apache.kafka.streams.processor.ProcessorContext; -import org.apache.kafka.streams.processor.internals.ProcessorContextImpl; -import org.apache.kafka.streams.processor.internals.StreamTask; -import org.springframework.util.ReflectionUtils; +import org.springframework.kafka.listener.ConsumerRecordRecoverer; +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; /** - * Custom implementation for {@link DeserializationExceptionHandler} that sends the - * records in error to a DLQ topic, then continue stream processing on new records. + * Custom implementation for {@link ConsumerRecordRecoverer} that keeps a collection of + * recoverer objects per input topics. These topics might be per input binding or multiplexed + * topics in a single binding. * * @author Soby Chacko * @since 2.0.0 */ -public class SendToDlqAndContinue implements DeserializationExceptionHandler { - - /** - * Key used for DLQ dispatchers. - */ - public static final String KAFKA_STREAMS_DLQ_DISPATCHERS = "spring.cloud.stream.kafka.streams.dlq.dispatchers"; +public class SendToDlqAndContinue implements ConsumerRecordRecoverer { /** * DLQ dispatcher per topic in the application context. The key here is not the actual * DLQ topic but the incoming topic that caused the error. */ - private Map dlqDispatchers = new HashMap<>(); + private Map dlqDispatchers = new HashMap<>(); /** * For a given topic, send the key/value record to DLQ topic. - * @param topic incoming topic that caused the error - * @param key to send - * @param value to send - * @param partition for the topic where this record should be sent + * + * @param consumerRecord consumer record + * @param exception exception */ - public void sendToDlq(String topic, byte[] key, byte[] value, int partition) { - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = this.dlqDispatchers.get(topic); - kafkaStreamsDlqDispatch.sendToDlq(key, value, partition); - } - - @Override - @SuppressWarnings("unchecked") - public DeserializationHandlerResponse handle(ProcessorContext context, - ConsumerRecord record, Exception exception) { - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch = this.dlqDispatchers - .get(record.topic()); - kafkaStreamsDlqDispatch.sendToDlq(record.key(), record.value(), - record.partition()); - context.commit(); - - // The following conditional block should be reconsidered when we have a solution - // for this SO problem: - // https://stackoverflow.com/questions/48470899/kafka-streams-deserialization-handler - // Currently it seems like when deserialization error happens, there is no commits - // happening and the - // following code will use reflection to get access to the underlying - // KafkaConsumer. - // It works with Kafka 1.0.0, but there is no guarantee it will work in future - // versions of kafka as - // we access private fields by name using reflection, but it is a temporary fix. - if (context instanceof ProcessorContextImpl) { - ProcessorContextImpl processorContextImpl = (ProcessorContextImpl) context; - Field task = ReflectionUtils.findField(ProcessorContextImpl.class, "task"); - ReflectionUtils.makeAccessible(task); - Object taskField = ReflectionUtils.getField(task, processorContextImpl); - - if (taskField.getClass().isAssignableFrom(StreamTask.class)) { - StreamTask streamTask = (StreamTask) taskField; - Field consumer = ReflectionUtils.findField(StreamTask.class, "consumer"); - ReflectionUtils.makeAccessible(consumer); - Object kafkaConsumerField = ReflectionUtils.getField(consumer, - streamTask); - if (kafkaConsumerField.getClass().isAssignableFrom(KafkaConsumer.class)) { - KafkaConsumer kafkaConsumer = (KafkaConsumer) kafkaConsumerField; - final Map consumedOffsetsAndMetadata = new HashMap<>(); - TopicPartition tp = new TopicPartition(record.topic(), - record.partition()); - OffsetAndMetadata oam = new OffsetAndMetadata(record.offset() + 1); - consumedOffsetsAndMetadata.put(tp, oam); - kafkaConsumer.commitSync(consumedOffsetsAndMetadata); - } - } - } - return DeserializationHandlerResponse.CONTINUE; - } - - @Override - @SuppressWarnings("unchecked") - public void configure(Map configs) { - this.dlqDispatchers = (Map) configs - .get(KAFKA_STREAMS_DLQ_DISPATCHERS); + public void sendToDlq(ConsumerRecord consumerRecord, Exception exception) { + DeadLetterPublishingRecoverer kafkaStreamsDlqDispatch = this.dlqDispatchers.get(consumerRecord.topic()); + kafkaStreamsDlqDispatch.accept(consumerRecord, exception); } void addKStreamDlqDispatch(String topic, - KafkaStreamsDlqDispatch kafkaStreamsDlqDispatch) { + DeadLetterPublishingRecoverer kafkaStreamsDlqDispatch) { this.dlqDispatchers.put(topic, kafkaStreamsDlqDispatch); } + @Override + public void accept(ConsumerRecord consumerRecord, Exception e) { + this.dlqDispatchers.get(consumerRecord.topic()).accept(consumerRecord, e); + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index cdda49a46..9047e3fc8 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -29,6 +29,9 @@ import java.util.stream.Stream; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.streams.kstream.GlobalKTable; +import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.KTable; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -80,7 +83,10 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, if (kafkaStreamMethod.isPresent()) { Method method = kafkaStreamMethod.get(); ResolvableType resolvableType = ResolvableType.forMethodReturnType(method, classObj); - resolvableTypeMap.put(key, resolvableType); + final Class rawClass = resolvableType.getGeneric(0).getRawClass(); + if (rawClass == KStream.class || rawClass == KTable.class || rawClass == GlobalKTable.class) { + resolvableTypeMap.put(key, resolvableType); + } } } catch (Exception e) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java index 2570952fb..c4d4450f3 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DeserializationErrorHandlerByKafkaTests.java @@ -154,7 +154,7 @@ public abstract class DeserializationErrorHandlerByKafkaTests { @Test @SuppressWarnings("unchecked") - public void test() { + public void test() throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); @@ -174,13 +174,11 @@ public abstract class DeserializationErrorHandlerByKafkaTests { embeddedKafka.consumeFromEmbeddedTopics(consumer1, "error.word1.groupx", "error.word2.groupx"); - // TODO: Investigate why the ordering matters below: i.e. - // if we consume from error.word1.groupx first, an exception is thrown. ConsumerRecord cr1 = KafkaTestUtils.getSingleRecord(consumer1, - "error.word2.groupx"); + "error.word1.groupx"); assertThat(cr1.value().equals("foobar")).isTrue(); ConsumerRecord cr2 = KafkaTestUtils.getSingleRecord(consumer1, - "error.word1.groupx"); + "error.word2.groupx"); assertThat(cr2.value().equals("foobar")).isTrue(); // Ensuring that the deserialization was indeed done by Kafka natively