diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java index a8f2c69c..15ae16fa 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * 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. @@ -20,6 +20,9 @@ import java.io.PrintWriter; import java.io.StringWriter; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; +import java.util.Optional; import java.util.function.BiConsumer; import java.util.function.BiFunction; @@ -34,7 +37,11 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.kafka.core.KafkaOperations; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.KafkaHeaders; +import org.springframework.kafka.support.serializer.DeserializationException; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * A {@link BiConsumer} that publishes a failed record to a dead-letter topic. @@ -47,8 +54,13 @@ public class DeadLetterPublishingRecoverer implements BiConsumer, Exception, TopicPartition> + DEFAULT_DESTINATION_RESOLVER = (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition()); + private final KafkaTemplate template; + private final Map, KafkaTemplate> templates; + private final boolean transactional; private final BiFunction, Exception, TopicPartition> destinationResolver; @@ -60,25 +72,65 @@ public class DeadLetterPublishingRecoverer implements BiConsumer template) { - this(template, (cr, e) -> new TopicPartition(cr.topic() + ".DLT", cr.partition())); + public DeadLetterPublishingRecoverer(KafkaTemplate template) { + this(template, DEFAULT_DESTINATION_RESOLVER); } /** * Create an instance with the provided template and destination resolving function, * that receives the failed consumer record and the exception and returns a - * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than 0, no - * partition is set when publishing to the topic. + * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than + * 0, no partition is set when publishing to the topic. * @param template the {@link KafkaTemplate} to use for publishing. * @param destinationResolver the resolving function. */ - public DeadLetterPublishingRecoverer(KafkaTemplate template, + public DeadLetterPublishingRecoverer(KafkaTemplate template, + BiFunction, Exception, TopicPartition> destinationResolver) { + this(Collections.singletonMap(Object.class, template), destinationResolver); + } + + /** + * Create an instance with the provided templates and a default destination resolving + * function that returns a TopicPartition based on the original topic (appended with + * ".DLT") from the failed record, and the same partition as the failed record. + * Therefore the dead-letter topic must have at least as many partitions as the + * original topic. The templates map keys are classes and the value the corresponding + * template to use for objects (producer record values) of that type. A + * {@link java.util.LinkedHashMap} is recommended when there is more than one + * template, to ensure the map is traversed in order. + * @param templates the {@link KafkaTemplate}s to use for publishing. + */ + public DeadLetterPublishingRecoverer(Map, KafkaTemplate> templates) { + this(templates, DEFAULT_DESTINATION_RESOLVER); + } + + /** + * Create an instance with the provided templates and destination resolving function, + * that receives the failed consumer record and the exception and returns a + * {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than + * 0, no partition is set when publishing to the topic. The templates map keys are + * classes and the value the corresponding template to use for objects (producer + * record values) of that type. A {@link java.util.LinkedHashMap} is recommended when + * there is more than one template, to ensure the map is traversed in order. + * @param templates the {@link KafkaTemplate}s to use for publishing. + * @param destinationResolver the resolving function. + */ + @SuppressWarnings("unchecked") + public DeadLetterPublishingRecoverer(Map, KafkaTemplate> templates, BiFunction, Exception, TopicPartition> destinationResolver) { - Assert.notNull(template, "The template cannot be null"); + Assert.isTrue(!ObjectUtils.isEmpty(templates), "At least one template is required"); Assert.notNull(destinationResolver, "The destinationResolver cannot be null"); - this.template = template; - this.transactional = template.isTransactional(); + this.template = templates.size() == 1 ? (KafkaTemplate) templates.values().iterator().next() : null; + this.templates = templates; + this.transactional = templates.values().iterator().next().isTransactional(); + Boolean tx = this.transactional; + Assert.isTrue(!templates.values() + .stream() + .map(t -> t.isTransactional()) + .filter(t -> !t.equals(tx)) + .findFirst() + .isPresent(), "All templates must have the same setting for transactional"); this.destinationResolver = destinationResolver; } @@ -87,42 +139,72 @@ public class DeadLetterPublishingRecoverer implements BiConsumer outRecord = createProducerRecord(record, tp, headers); - if (this.transactional && !this.template.inTransaction()) { - this.template.executeInTransaction(t -> { + DeserializationException deserEx = ListenerUtils.getExceptionFromHeader(record, + ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER, logger); + if (deserEx == null) { + deserEx = ListenerUtils.getExceptionFromHeader(record, + ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER, logger); + } + ProducerRecord outRecord = createProducerRecord(record, tp, headers, + deserEx == null ? null : deserEx.getData()); + KafkaTemplate kafkaTemplate = findTemplateForValue(outRecord.value()); + if (this.transactional && !kafkaTemplate.inTransaction()) { + kafkaTemplate.executeInTransaction(t -> { publish(outRecord, t); return null; }); } else { - publish(outRecord, this.template); + publish(outRecord, kafkaTemplate); } } + @SuppressWarnings("unchecked") + private KafkaTemplate findTemplateForValue(Object value) { + if (this.template != null) { + return this.template; + } + Optional> key = this.templates.keySet() + .stream() + .filter((k) -> k.isAssignableFrom(value.getClass())) + .findFirst(); + if (key.isPresent()) { + return (KafkaTemplate) this.templates.get(key.get()); + } + if (logger.isWarnEnabled()) { + logger.warn("Failed to find a template for " + value.getClass() + " attemting to use the last entry"); + } + return (KafkaTemplate) this.templates.values() + .stream() + .reduce((first, second) -> second) + .get(); + } + /** - * Subclasses can override this method to customize the producer record to send to the DLQ. - * The default implementation simply copies the key and value from the consumer record - * and adds the headers. The timestamp is not set (the original timestamp is in one of - * the headers). - * IMPORTANT: if the partition in the {@link TopicPartition} is less than 0, it must be set to null - * in the {@link ProducerRecord}. + * Subclasses can override this method to customize the producer record to send to the + * DLQ. The default implementation simply copies the key and value from the consumer + * record and adds the headers. The timestamp is not set (the original timestamp is in + * one of the headers). IMPORTANT: if the partition in the {@link TopicPartition} is + * less than 0, it must be set to null in the {@link ProducerRecord}. * @param record the failed record - * @param topicPartition the {@link TopicPartition} returned by the destination resolver. + * @param topicPartition the {@link TopicPartition} returned by the destination + * resolver. * @param headers the headers - original record headers plus DLT headers. + * @param value the value to use instead of the consumer record value. * @return the producer record to send. * @see KafkaHeaders */ protected ProducerRecord createProducerRecord(ConsumerRecord record, - TopicPartition topicPartition, RecordHeaders headers) { + TopicPartition topicPartition, RecordHeaders headers, @Nullable byte[] value) { return new ProducerRecord<>(topicPartition.topic(), topicPartition.partition() < 0 ? null : topicPartition.partition(), - record.key(), record.value(), headers); + record.key(), value == null ? record.value() : value, headers); } - private void publish(ProducerRecord outRecord, KafkaOperations template) { + private void publish(ProducerRecord outRecord, KafkaOperations kafkaTemplate) { try { - template.send(outRecord).addCallback(result -> { + kafkaTemplate.send(outRecord).addCallback(result -> { if (logger.isDebugEnabled()) { logger.debug("Successful dead-letter publication: " + result); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 6ca9fdfd..8bb75d35 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -16,9 +16,6 @@ package org.springframework.kafka.listener; -import java.io.ByteArrayInputStream; -import java.io.IOException; -import java.io.ObjectInputStream; import java.lang.reflect.Type; import java.time.Duration; import java.util.ArrayList; @@ -54,9 +51,6 @@ import org.apache.kafka.common.Metric; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.Headers; -import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.kafka.KafkaException; @@ -1341,21 +1335,9 @@ public class KafkaMessageListenerContainer // NOSONAR line count } public void checkDeser(final ConsumerRecord record, String headerName) { - Header header = record.headers().lastHeader(headerName); - if (header != null) { - try { - DeserializationException ex = (DeserializationException) new ObjectInputStream( - new ByteArrayInputStream(header.value())).readObject(); - Headers headers = new RecordHeaders(Arrays.stream(record.headers().toArray()) - .filter(h -> !h.key() - .startsWith(ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX)) - .collect(Collectors.toList())); - ex.setHeaders(headers); - throw ex; - } - catch (IOException | ClassNotFoundException | ClassCastException e) { - this.logger.error("Failed to deserialize a deserialization exception", e); - } + DeserializationException exception = ListenerUtils.getExceptionFromHeader(record, headerName, this.logger); + if (exception != null) { + throw exception; } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java index a3685218..a4d50561 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ListenerUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-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. @@ -16,6 +16,21 @@ package org.springframework.kafka.listener; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.util.Arrays; +import java.util.stream.Collectors; + +import org.apache.commons.logging.Log; +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.RecordHeaders; + +import org.springframework.kafka.support.serializer.DeserializationException; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -55,4 +70,36 @@ public final class ListenerUtils { return listenerType; } + /** + * Extract a {@link DeserializationException} from the supplied header name, if + * present. + * @param record the consumer record. + * @param headerName the header name. + * @param logger the logger for logging errors. + * @return the exception or null. + * @since 2.3 + */ + @Nullable + public static DeserializationException getExceptionFromHeader(final ConsumerRecord record, + String headerName, Log logger) { + + Header header = record.headers().lastHeader(headerName); + if (header != null) { + try { + DeserializationException ex = (DeserializationException) new ObjectInputStream( + new ByteArrayInputStream(header.value())).readObject(); + Headers headers = new RecordHeaders(Arrays.stream(record.headers().toArray()) + .filter(h -> !h.key() + .startsWith(ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX)) + .collect(Collectors.toList())); + ex.setHeaders(headers); + return ex; + } + catch (IOException | ClassNotFoundException | ClassCastException e) { + logger.error("Failed to deserialize a deserialization exception", e); + } + } + return null; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java index e0673a49..4c24daa1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/SeekToCurrentErrorHandler.java @@ -241,7 +241,9 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler { } } - private BiPredicate, Exception> getSkipPredicate(List> records, Exception thrownException) { + private BiPredicate, Exception> getSkipPredicate(List> records, + Exception thrownException) { + if (this.classifier.classify(thrownException)) { return this.failureTracker::skip; } @@ -263,7 +265,9 @@ public class SeekToCurrentErrorHandler implements ContainerAwareErrorHandler { classified.put(MethodArgumentResolutionException.class, false); classified.put(NoSuchMethodException.class, false); classified.put(ClassCastException.class, false); - return new ExtendedBinaryExceptionClassifier(classified, true); + ExtendedBinaryExceptionClassifier defaultClassifier = new ExtendedBinaryExceptionClassifier(classified, true); + defaultClassifier.setTraverseCauses(true); + return defaultClassifier; } /** diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java index c76bbd88..dae49fcc 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java @@ -31,6 +31,7 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -45,6 +46,8 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.ByteArraySerializer; +import org.apache.kafka.common.serialization.Serializer; import org.junit.ClassRule; import org.junit.Test; @@ -54,6 +57,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.event.ConsumerStoppedEvent; import org.springframework.kafka.listener.ContainerProperties.AckMode; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; +import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -79,7 +84,8 @@ public class SeekToCurrentRecovererTests { Map props = KafkaTestUtils.consumerProps("seekTestMaxFailures", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props, null, + new ErrorHandlingDeserializer2<>(new JsonDeserializer<>(String.class))); ContainerProperties containerProps = new ContainerProperties(topic1); containerProps.setPollTimeout(10_000); @@ -87,6 +93,10 @@ public class SeekToCurrentRecovererTests { senderProps.put(ProducerConfig.RETRIES_CONFIG, 1); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); final KafkaTemplate template = new KafkaTemplate<>(pf); + Serializer byteArraySerializer = new ByteArraySerializer(); + @SuppressWarnings("unchecked") + DefaultKafkaProducerFactory dltPf = + new DefaultKafkaProducerFactory(senderProps, null, (Serializer) byteArraySerializer); final CountDownLatch latch = new CountDownLatch(1); AtomicReference data = new AtomicReference<>(); containerProps.setMessageListener((MessageListener) message -> { @@ -102,8 +112,12 @@ public class SeekToCurrentRecovererTests { container.setBeanName("testSeekMaxFailures"); final CountDownLatch recoverLatch = new CountDownLatch(1); final AtomicReference failedGroupId = new AtomicReference<>(); - DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template, - (r, e) -> new TopicPartition(topic1DLT, r.partition())) { + Map, KafkaTemplate> templates = new LinkedHashMap<>(); + templates.put(String.class, template); + templates.put(byte[].class, new KafkaTemplate<>(dltPf)); + DeadLetterPublishingRecoverer recoverer = + new DeadLetterPublishingRecoverer(templates, + (r, e) -> new TopicPartition(topic1DLT, r.partition())) { @Override public void accept(ConsumerRecord record, Exception exception) { @@ -126,19 +140,28 @@ public class SeekToCurrentRecovererTests { container.start(); template.setDefaultTopic(topic1); - template.sendDefault(0, 0, "foo"); - template.sendDefault(0, 0, "bar"); + template.sendDefault(0, 0, "\"foo\""); + template.sendDefault(0, 0, "\"bar\""); assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); assertThat(data.get()).isEqualTo("bar"); assertThat(recoverLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(failedGroupId.get()).isEqualTo("seekTestMaxFailures"); - container.stop(); - Consumer consumer = cf.createConsumer(); + + props.put(ConsumerConfig.GROUP_ID_CONFIG, "seekTestMaxFailures.dlt"); + DefaultKafkaConsumerFactory dltcf = new DefaultKafkaConsumerFactory<>(props); + Consumer consumer = dltcf.createConsumer(); embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic1DLT); ConsumerRecord dltRecord = KafkaTestUtils.getSingleRecord(consumer, topic1DLT); assertThat(dltRecord.value()).isEqualTo("foo"); + template.sendDefault(0, 0, "junkJson"); + dltRecord = KafkaTestUtils.getSingleRecord(consumer, topic1DLT); + assertThat(dltRecord.value()).isEqualTo("junkJson"); + container.stop(); pf.destroy(); + dltPf.destroy(); + consumer.close(); assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(errorHandler, times(4)).handle(any(), any(), any(), any()); verify(errorHandler).clearThreadState(); } @@ -155,7 +178,7 @@ public class SeekToCurrentRecovererTests { eh.handle(new RuntimeException(), records, consumer, null); fail("Expected exception"); } - catch (KafkaException e) { + catch (@SuppressWarnings("unused") KafkaException e) { // NOSONAR } verify(consumer).seek(new TopicPartition("foo", 0), 0L); @@ -197,7 +220,7 @@ public class SeekToCurrentRecovererTests { eh.handle(new RuntimeException(), records, consumer, container); fail("Expected exception"); } - catch (KafkaException e) { + catch (@SuppressWarnings("unused") KafkaException e) { // NOSONAR } verify(consumer).seek(new TopicPartition("foo", 0), 0L); @@ -234,7 +257,7 @@ public class SeekToCurrentRecovererTests { eh.handle(new RuntimeException(), records, consumer, null); fail("Expected exception"); } - catch (KafkaException e) { + catch (@SuppressWarnings("unused") KafkaException e) { // NOSONAR } } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 268e0937..eaaf12b6 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -3084,6 +3084,28 @@ The record sent to the dead-letter topic is enhanced with the following headers: * `KafkaHeaders.DLT_ORIGINAL_TIMESTAMP_TYPE`: The original timestamp type. +Starting with version 2.3, when used in conjunction with an `ErrorHandlingDeserializer2`, the publisher will restore the record `value()`, in the dead-letter producer record, to the original value that failed to be deserialized. +Previously, the `value()` was null and user code had to decode the `DeserializationException` from the message headers. +In addition, you can provide multiple `KafkaTemplate` s to the publisher; this might be needed, for example, if you want to publish the `byte[]` from a `DeserializationException`, as well as values using a different serializer from records that were deserialized successfully. +Here is an example of configuring the publisher with `KafkaTemplate` s that use a `String` and `byte[]` serializer: + +==== +[source, java] +---- +@Bean +public DeadLetterPublishingRecoverer publisher(KafkaTemplate stringTemplate, + KafkaTemplate bytesTemplate) { + + Map, KafkaTemplate> templates = new LinkedHashMap<>(); + templates.put(String.class, stringTemplate); + templates.put(byte[].class, bytesTemplate); + return new DeadLetterPublishingRecoverer(templates); +} +---- +==== + +The publisher uses the map keys to locate a template that is suitable for the `value()` about to be published. +A `LinkedHashMap` is recommended so that the keys are examined in order. [[kerberos]] ==== Kerberos diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 848ddf3a..73cb9de0 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -21,6 +21,10 @@ It now sets it to false automatically unless specifically set in the consumer fa The `SeekToCurrentErrorHandler` now treats certain exceptions as fatal and disables retry for those, invoking the recoverer on first failure. See <> for more information. +The `DeadLetterPublishingRecoverer`, when used in conjunction with an `ErrorHandlingDeserializer2`, now sets the payload of the message sent to the dead-letter topic, to the original value that could not be deserialized. +Previously, it was `null` and user code needed to extract the `DeserializationException` from the message headers. +See <> for more information. + ==== TopicBuilder A new class `TopicBuilder` is provided for more convenient creation of `NewTopic` `@Bean` s for automatic topic provisioning.