From 65dd706a6ad6e4e517736f59f975d8c7cf3b8b7c Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 2 Oct 2019 22:46:22 -0400 Subject: [PATCH] Kafka Streams DLQ enhancements Use DeadLetterPublishingRecoverer from Spring Kafka instead of custom DLQ components in the binder. Remove code that is no longer needed for DLQ purposes. In Kafka Streams, always set group to application id if the user doesn't set an explicit group. Upgrade Spring Kafka to 2.3.0 and SIK to 3.2.0 Resolves #761 --- docs/src/main/asciidoc/kafka-streams.adoc | 56 +------ pom.xml | 4 +- .../AbstractKafkaStreamsBinderProcessor.java | 15 +- .../kafka/streams/GlobalKTableBinder.java | 13 +- .../GlobalKTableBinderConfiguration.java | 4 +- .../binder/kafka/streams/KStreamBinder.java | 13 +- .../streams/KStreamBinderConfiguration.java | 9 +- .../binder/kafka/streams/KTableBinder.java | 13 +- .../streams/KTableBinderConfiguration.java | 4 +- ...StreamsBinderSupportAutoConfiguration.java | 16 +- .../streams/KafkaStreamsBinderUtils.java | 83 ++++++++-- .../streams/KafkaStreamsDlqDispatch.java | 152 ------------------ ...KafkaStreamsMessageConversionDelegate.java | 21 ++- .../kafka/streams/SendToDlqAndContinue.java | 97 +++-------- ...KafkaStreamsFunctionBeanPostProcessor.java | 8 +- ...serializationErrorHandlerByKafkaTests.java | 8 +- 16 files changed, 137 insertions(+), 379 deletions(-) delete mode 100644 spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java 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