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 3512abd0..18c8ce9d 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,6 +16,9 @@ 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; @@ -50,6 +53,9 @@ 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; @@ -70,6 +76,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; import org.springframework.kafka.support.TransactionSupport; import org.springframework.kafka.support.serializer.DeserializationException; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; import org.springframework.kafka.transaction.KafkaAwareTransactionManager; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.TaskScheduler; @@ -417,6 +424,10 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final Duration pollTimeout = Duration.ofMillis(this.containerProperties.getPollTimeout()); + private final boolean checkNullKeyForExceptions; + + private final boolean checkNullValueForExceptions; + private volatile Map definedPartitions; private volatile Collection assignedPartitions; @@ -521,6 +532,18 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (this.containerProperties.isLogContainerConfig()) { this.logger.info(this); } + Map props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties(); + this.checkNullKeyForExceptions = checkDeserializer(props.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG)); + this.checkNullValueForExceptions = checkDeserializer( + props.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG)); + } + + private boolean checkDeserializer(Object deser) { + return deser instanceof Class + ? ((Class) deser).equals(ErrorHandlingDeserializer2.class) + : deser instanceof String + ? ((String) deser).equals(ErrorHandlingDeserializer2.class.getName()) + : false; } protected void checkConsumer() { @@ -1136,6 +1159,12 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (record.key() instanceof DeserializationException) { throw (DeserializationException) record.key(); } + if (record.value() == null && this.checkNullValueForExceptions) { + checkDeser(record, ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER); + } + if (record.key() == null && this.checkNullKeyForExceptions) { + checkDeser(record, ErrorHandlingDeserializer2.KEY_DESERIALIZER_EXCEPTION_HEADER); + } switch (this.listenerType) { case ACKNOWLEDGING_CONSUMER_AWARE: this.listener.onMessage(record, @@ -1197,6 +1226,25 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener return null; } + 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); + } + } + } + public void ackCurrent(final ConsumerRecord record, @SuppressWarnings("rawtypes") Producer producer) { if (this.isRecordAck) { Map offsetsToCommit = diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DeserializationException.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DeserializationException.java index 91fb4587..21332ce2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DeserializationException.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/DeserializationException.java @@ -35,14 +35,14 @@ import org.springframework.lang.Nullable; public class DeserializationException extends KafkaException { @Nullable - private final Headers headers; + private Headers headers; private final byte[] data; private final boolean isKey; public DeserializationException(String message, byte[] data, boolean isKey, Throwable cause) { - this(message, null, data, isKey, cause); // NOSONAR test coverage + this(message, null, data, isKey, cause); } public DeserializationException(String message, @Nullable Headers headers, byte[] data, // NOSONAR array reference @@ -59,6 +59,10 @@ public class DeserializationException extends KafkaException { return this.headers; } + public void setHeaders(@Nullable Headers headers) { + this.headers = headers; + } + public byte[] getData() { return this.data; // NOSONAR array reference } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java index 68ff1f82..439dd9cb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer.java @@ -34,10 +34,12 @@ import org.springframework.util.ClassUtils; * * @author Gary Russell * @author Artem Bilan + * @deprecated in favor of {@link ErrorHandlingDeserializer2}. * * @since 2.2 * */ +@Deprecated public class ErrorHandlingDeserializer implements ExtendedDeserializer { /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer2.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer2.java new file mode 100644 index 00000000..d97287fd --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer2.java @@ -0,0 +1,200 @@ +/* + * Copyright 2018 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 + * + * http://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.kafka.support.serializer; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.util.Map; +import java.util.function.BiFunction; + +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.ExtendedDeserializer; + +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +/** + * Delegating key/value deserializer that catches exceptions, returning them + * in the headers as serialized java objects. + * + * @param class of the entity, representing messages + * + * @author Gary Russell + * @author Artem Bilan + * + * @since 2.2 + * + */ +public class ErrorHandlingDeserializer2 implements ExtendedDeserializer { + + /** + * Header name for deserialization exceptions. + */ + public static final String KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX = "springDeserializerException"; + + /** + * Header name for deserialization exceptions. + */ + public static final String KEY_DESERIALIZER_EXCEPTION_HEADER = KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX + "Key"; + + /** + * Heaader name for deserialization exceptions. + */ + public static final String VALUE_DESERIALIZER_EXCEPTION_HEADER = KEY_DESERIALIZER_EXCEPTION_HEADER_PREFIX + "Value"; + + /** + * Supplier for a T when deserialization fails. + */ + public static final String KEY_FUNCTION = "spring.deserializer.key.function"; + + /** + * Supplier for a T when deserialization fails. + */ + public static final String VALUE_FUNCTION = "spring.deserializer.value.function"; + + /** + * Property name for the delegate key deserializer. + */ + public static final String KEY_DESERIALIZER_CLASS = "spring.deserializer.key.delegate.class"; + + /** + * Property name for the delegate value deserializer. + */ + public static final String VALUE_DESERIALIZER_CLASS = "spring.deserializer.value.delegate.class"; + + private ExtendedDeserializer delegate; + + private boolean isKey; + + private BiFunction failedDeserializationFunction; + + public ErrorHandlingDeserializer2() { + super(); + } + + public ErrorHandlingDeserializer2(Deserializer delegate) { + this.delegate = setupDelegate(delegate); + } + + public void setFailedDeserializationFunction(BiFunction failedDeserializationFunction) { + this.failedDeserializationFunction = failedDeserializationFunction; + } + + + @Override + public void configure(Map configs, boolean isKey) { + setupDelegate(configs, isKey ? KEY_DESERIALIZER_CLASS : VALUE_DESERIALIZER_CLASS); + Assert.state(this.delegate != null, "No delegate deserializer configured"); + this.delegate.configure(configs, isKey); + this.isKey = isKey; + setupFunction(configs, isKey ? KEY_FUNCTION : VALUE_FUNCTION); + } + + public void setupDelegate(Map configs, String configKey) { + if (configs.containsKey(configKey)) { + try { + Object value = configs.get(configKey); + Class clazz = value instanceof Class ? (Class) value : ClassUtils.forName((String) value, null); + this.delegate = setupDelegate(clazz.newInstance()); + } + catch (ClassNotFoundException | LinkageError | InstantiationException | IllegalAccessException e) { + throw new IllegalStateException(e); + } + } + } + + @SuppressWarnings("unchecked") + private ExtendedDeserializer setupDelegate(Object delegate) { + Assert.isInstanceOf(Deserializer.class, delegate, "'delegate' must be a 'Deserializer', not a "); + return delegate instanceof ExtendedDeserializer + ? (ExtendedDeserializer) delegate + : ExtendedDeserializer.Wrapper.ensureExtended((Deserializer) delegate); + } + + @SuppressWarnings("unchecked") + private void setupFunction(Map configs, String configKey) { + if (configs.containsKey(configKey)) { + try { + Object value = configs.get(configKey); + Class clazz = value instanceof Class ? (Class) value : ClassUtils.forName((String) value, null); + Assert.isTrue(BiFunction.class.isAssignableFrom(clazz), "'function' must be a 'BiFunction ', not a " + + clazz.getName()); + this.failedDeserializationFunction = (BiFunction) clazz.newInstance(); + } + catch (ClassNotFoundException | LinkageError | InstantiationException | IllegalAccessException e) { + throw new IllegalStateException(e); + } + } + } + + @Override + public T deserialize(String topic, byte[] data) { + try { + return this.delegate.deserialize(topic, data); + } + catch (Exception e) { + return this.failedDeserializationFunction != null + ? this.failedDeserializationFunction.apply(data, null) + : null; + } + } + + @Override + public T deserialize(String topic, Headers headers, byte[] data) { + try { + return this.delegate.deserialize(topic, headers, data); + } + catch (Exception e) { + deserializationException(headers, data, e); + return this.failedDeserializationFunction != null + ? this.failedDeserializationFunction.apply(data, headers) + : null; + } + } + + @Override + public void close() { + this.delegate.close(); + } + + private void deserializationException(Headers headers, byte[] data, Exception e) { + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + DeserializationException exception = new DeserializationException("failed to deserialize", data, this.isKey, e); + try { + new ObjectOutputStream(stream).writeObject(exception); + } + catch (IOException ex) { + try { + exception = new DeserializationException("failed to deserialize", + data, this.isKey, new RuntimeException("Could not deserialize type " + + e.getClass().getName() + " with message " + e.getMessage() + + " failure: " + ex.getMessage())); + new ObjectOutputStream(stream).writeObject(exception); + } + catch (IOException ex2) { + throw new IllegalStateException("Could not serialize a DeserializationException", ex2); + } + } + headers.add( + new RecordHeader(this.isKey ? KEY_DESERIALIZER_EXCEPTION_HEADER : VALUE_DESERIALIZER_EXCEPTION_HEADER, + stream.toByteArray())); + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java new file mode 100644 index 00000000..f570fed7 --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java @@ -0,0 +1,221 @@ +/* + * Copyright 2018 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 + * + * http://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.kafka.annotation; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BiFunction; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.support.converter.BytesJsonMessageConverter; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * + * @since 2.1.1 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext +public class BatchListenerConversion2Tests { + + private static final String DEFAULT_TEST_GROUP_ID = "blc2"; + + @ClassRule // one topic to preserve order + public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 1, "blc.2.1"); + + @Autowired + private Config config; + + @Autowired + private KafkaTemplate template; + + @Test + public void testBatchOfPojosWithABadOne() throws Exception { + Listener listener = this.config.listener1(); + String topic = "blc.2.1"; + this.template.send(topic, "{\"bar\":\"baz\"}"); + this.template.send(topic, "junk"); + this.template.send(topic, "{\"bar\":\"baz\"}"); + assertThat(listener.latch1.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(listener.badFoo).isInstanceOf(BadFoo.class); + assertThat(listener.receivedFoos).isEqualTo(2); + } + + @Configuration + @EnableKafka + public static class Config { + + @Bean + public KafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory()); + factory.setBatchListener(true); + factory.setReplyTemplate(template()); + return factory; + } + + @Bean + public DefaultKafkaConsumerFactory consumerFactory() { + return new DefaultKafkaConsumerFactory<>(consumerConfigs()); + } + + @Bean + public Map consumerConfigs() { + Map consumerProps = + KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka.getEmbeddedKafka()); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); + consumerProps.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class); + consumerProps.put(ErrorHandlingDeserializer2.VALUE_FUNCTION, FailedFooProvider.class); + consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Foo.class.getName()); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return consumerProps; + } + + @Bean + public KafkaTemplate template() { + KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFactory()); + return kafkaTemplate; + } + + @Bean + public BytesJsonMessageConverter converter() { + return new BytesJsonMessageConverter(); + } + + @Bean + public ProducerFactory producerFactory() { + return new DefaultKafkaProducerFactory<>(producerConfigs()); + } + + @Bean + public Map producerConfigs() { + Map props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka()); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + return props; + } + + @Bean + public Listener listener1() { + return new Listener(); + } + + } + + public static class Listener { + + private final CountDownLatch latch1 = new CountDownLatch(3); + + private volatile Foo badFoo; + + private volatile int receivedFoos; + + @KafkaListener(id = "deser", topics = "blc.2.1") + public void listen1(List foos) { + foos.forEach(f -> { + if (f.getBar() == null) { + this.badFoo = f; + } + else { + this.receivedFoos++; + } + this.latch1.countDown(); + }); + } + + } + + public static class Foo { + + private String bar; + + public Foo() { + super(); + } + + public Foo(String bar) { + this.bar = bar; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public String toString() { + return "Foo [bar=" + this.bar + "]"; + } + + } + + public static class BadFoo extends Foo { + + private final byte[] failedDecode; + + public BadFoo(byte[] failedDecode) { + this.failedDecode = failedDecode; + } + + public byte[] getFailedDecode() { + return this.failedDecode; + } + + } + + public static class FailedFooProvider implements BiFunction { + + @Override + public Foo apply(byte[] t, Headers u) { + return new BadFoo(t); + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java index fb85dc63..2d0e2ccd 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java @@ -18,6 +18,8 @@ package org.springframework.kafka.listener; import static org.assertj.core.api.Assertions.assertThat; +import java.io.ByteArrayInputStream; +import java.io.ObjectInputStream; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -25,7 +27,9 @@ import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.header.Header; import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.ExtendedDeserializer; import org.apache.kafka.common.serialization.StringDeserializer; @@ -44,7 +48,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.support.serializer.DeserializationException; -import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.test.annotation.DirtiesContext; @@ -77,11 +81,11 @@ public class ErrorHandlingDeserializerTests { } @Test - public void unitTests() { - ErrorHandlingDeserializer ehd = new ErrorHandlingDeserializer<>(new StringDeserializer()); + public void unitTests() throws Exception { + ErrorHandlingDeserializer2 ehd = new ErrorHandlingDeserializer2<>(new StringDeserializer()); assertThat(ehd.deserialize("topic", "foo".getBytes())).isEqualTo("foo"); ehd.close(); - ehd = new ErrorHandlingDeserializer<>(new Deserializer() { + ehd = new ErrorHandlingDeserializer2<>(new Deserializer() { @Override public void configure(Map configs, boolean isKey) { @@ -97,8 +101,11 @@ public class ErrorHandlingDeserializerTests { } }); - Object result = ehd.deserialize("topic", "foo".getBytes()); - assertThat(result).isInstanceOf(DeserializationException.class); + Headers headers = new RecordHeaders(); + Object result = ehd.deserialize("topic", headers, "foo".getBytes()); + assertThat(result).isNull(); + Header deser = headers.lastHeader(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_EXCEPTION_HEADER); + assertThat(new ObjectInputStream(new ByteArrayInputStream(deser.value())).readObject()).isInstanceOf(DeserializationException.class); ehd.close(); } @@ -122,7 +129,6 @@ public class ErrorHandlingDeserializerTests { this.latch.countDown(); } - @Bean public EmbeddedKafkaBroker embeddedKafka() { return new EmbeddedKafkaBroker(1, true, 1, TOPIC); @@ -134,11 +140,11 @@ public class ErrorHandlingDeserializerTests { new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(cf()); factory.setErrorHandler((t, r) -> { - if (r.value() instanceof DeserializationException) { + if (r.value() == null && t instanceof DeserializationException) { this.valueErrorCount++; - this.headers = ((DeserializationException) r.value()).getHeaders(); + this.headers = ((DeserializationException) t).getHeaders(); } - else if (r.key() instanceof DeserializationException) { + else if (r.key() == null && t instanceof DeserializationException) { this.keyErrorCount++; } this.latch.countDown(); @@ -150,10 +156,10 @@ public class ErrorHandlingDeserializerTests { public ConsumerFactory cf() { Map props = KafkaTestUtils.consumerProps(TOPIC, "false", embeddedKafka()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); - props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); - props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class); - props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, FailSometimesDeserializer.class.getName()); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); + props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); + props.put(ErrorHandlingDeserializer2.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class); + props.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, FailSometimesDeserializer.class.getName()); return new DefaultKafkaConsumerFactory<>(props); } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 32bb8575..89921278 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -84,7 +84,7 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.TopicPartitionInitialOffset; import org.springframework.kafka.support.TopicPartitionInitialOffset.SeekPosition; -import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer; +import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.EmbeddedKafkaBroker; @@ -1696,8 +1696,8 @@ public class KafkaMessageListenerContainerTests { this.logger.info("Start JSON4"); Map props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka); - ErrorHandlingDeserializer errorHandlingDeserializer = - new ErrorHandlingDeserializer<>(new JsonDeserializer<>(Foo1.class, false)); + ErrorHandlingDeserializer2 errorHandlingDeserializer = + new ErrorHandlingDeserializer2<>(new JsonDeserializer<>(Foo1.class, false)); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props, new IntegerDeserializer(), errorHandlingDeserializer); diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index 05fa5cc4..1c707c3b 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -7,4 +7,5 @@ + diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 6dd67e0b..9735c95e 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1937,22 +1937,28 @@ Generally, the `BytesJsonMessageConverter` is more efficient because it avoids a ===== ErrorHandlingDeserializer When a deserializer fails to deserialize a message, Spring has no way to handle the problem because it occurs before the `poll()` returns. -To solve this problem, version 2.2 introduced the `ErrorHandlingDeserializer`. +To solve this problem, version 2.2 introduced the `ErrorHandlingDeserializer2`. This deserializer delegates to a real deserializer (key or value). -If the delegate fails to deserialize the record content, the `ErrorHandlingDeserializer` returns a `DeserializationException` instead, containing the cause and raw bytes. -When using a record-level `MessageListener`, if either the key or value contains a `DeserializationException`, the container's `ErrorHandler` is called with the failed `ConsumerRecord`. -When using a `BatchMessageListener`, the failed record is passed to the application along with the remaining records in the batch, so it is the responsibility of the application listener to check whether the key or value in a particular record is a `DeserializationException`. +If the delegate fails to deserialize the record content, the `ErrorHandlingDeserializer2` returns a `null` value and a `DeserializationException` in a header, containing the cause and raw bytes. +When using a record-level `MessageListener`, if either the key or value contains a `DeserializationException` header, the container's `ErrorHandler` is called with the failed `ConsumerRecord`; the record is not passed to the listener. -You can use the `DefaultKafkaConsumerFactory` constructor that takes key and value `Deserializer` objects and wire in appropriate `ErrorHandlingDeserializer` configured with the proper delegates. +Alternatively, you can configure a `failedDeserializationFunction` which is a `BiConsumer`. +This function is invoked to create an instance of `T` which is passed to the listener, as normal. +The raw record value and headers are provided to the function. +The `DeserializationException` can be found (as a serialized Java object) in headers; see the javadocs for the `ErrorHandlingDeserializer2` for more information. + +When using a `BatchMessageListener`, you **must** provide a `failedDeserializationFunction`, otherwise, the batch of records will not be type safe. + +You can use the `DefaultKafkaConsumerFactory` constructor that takes key and value `Deserializer` objects and wire in appropriate `ErrorHandlingDeserializer2` configured with the proper delegates. Alternatively, you can use consumer configuration properties which are used by the `ErrorHandlingDeserializer` to instantiate the delegates. -The property names are `ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS` and `ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS`; the property value can be a class or class name. +The property names are `ErrorHandlingDeserializer2.KEY_DESERIALIZER_CLASS` and `ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS`; the property value can be a class or class name. For example: [source, java] ---- ... // other props -props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); -props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class); +props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); +props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, JsonDeserializer.class); props.put(JsonDeserializer.KEY_DEFAULT_TYPE, "com.example.MyKey") props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class.getName()); @@ -1961,6 +1967,45 @@ props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example") return new DefaultKafkaConsumerFactory<>(props); ---- +The following is an example of using a `failedDeserializationFunction`. + +[source, java] +---- +public class BadFoo extends Foo { + + private final byte[] failedDecode; + + public BadFoo(byte[] failedDecode) { + this.failedDecode = failedDecode; + } + + public byte[] getFailedDecode() { + return this.failedDecode; + } + +} + +public class FailedFooProvider implements BiFunction { + + @Override + public Foo apply(byte[] t, Headers u) { + return new BadFoo(t); + } + +} +---- + +and config + +[source, java] +---- +... +consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class); +consumerProps.put(ErrorHandlingDeserializer2.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class); +consumerProps.put(ErrorHandlingDeserializer2.VALUE_FUNCTION, FailedFooProvider.class); +... +---- + [[payload-conversion-with-batch]] ===== Payload Conversion with Batch Listeners