diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java index af937e78..5e4c1ddc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java @@ -39,6 +39,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Adrian Chlebosz * @author Antonin Arquey + * @author Dan Blackney * @since 2.8 * */ @@ -181,6 +182,19 @@ public class CommonDelegatingErrorHandler implements CommonErrorHandler { } } + @Override + public boolean handleOne(Exception thrownException, ConsumerRecord record, Consumer consumer, + MessageListenerContainer container) { + + CommonErrorHandler handler = findDelegate(thrownException); + if (handler != null) { + return handler.handleOne(thrownException, record, consumer, container); + } + else { + return this.defaultErrorHandler.handleOne(thrownException, record, consumer, container); + } + } + @Nullable private CommonErrorHandler findDelegate(Throwable thrownException) { Throwable cause = findCause(thrownException); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/CommonDelegatingErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/CommonDelegatingErrorHandlerTests.java index c58b25da..225f5b43 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/CommonDelegatingErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/CommonDelegatingErrorHandlerTests.java @@ -19,6 +19,7 @@ package org.springframework.kafka.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -29,6 +30,7 @@ import java.util.Collections; import java.util.Map; import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.junit.jupiter.api.Test; @@ -42,13 +44,14 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; * @author Gary Russell * @author Adrian Chlebosz * @author Antonin Arquey + * @author Dan Blackney * @since 2.8 * */ public class CommonDelegatingErrorHandlerTests { @Test - void testRecordDelegates() { + void testHandleRemainingDelegates() { var def = mock(CommonErrorHandler.class); var one = mock(CommonErrorHandler.class); var two = mock(CommonErrorHandler.class); @@ -72,7 +75,7 @@ public class CommonDelegatingErrorHandlerTests { } @Test - void testBatchDelegates() { + void testHandleBatchDelegates() { var def = mock(CommonErrorHandler.class); var one = mock(CommonErrorHandler.class); var two = mock(CommonErrorHandler.class); @@ -95,6 +98,54 @@ public class CommonDelegatingErrorHandlerTests { verify(one).handleBatch(any(), any(), any(), any(), any()); } + @Test + void testHandleOtherExceptionDelegates() { + var def = mock(CommonErrorHandler.class); + var one = mock(CommonErrorHandler.class); + var two = mock(CommonErrorHandler.class); + var three = mock(CommonErrorHandler.class); + var eh = new CommonDelegatingErrorHandler(def); + eh.setErrorHandlers(Map.of(IllegalStateException.class, one, IllegalArgumentException.class, two)); + eh.addDelegate(RuntimeException.class, three); + + eh.handleOtherException(wrap(new IOException()), mock(Consumer.class), + mock(MessageListenerContainer.class), true); + verify(def).handleOtherException(any(), any(), any(), anyBoolean()); + eh.handleOtherException(wrap(new KafkaException("test")), mock(Consumer.class), + mock(MessageListenerContainer.class), true); + verify(three).handleOtherException(any(), any(), any(), anyBoolean()); + eh.handleOtherException(wrap(new IllegalArgumentException()), mock(Consumer.class), + mock(MessageListenerContainer.class), true); + verify(two).handleOtherException(any(), any(), any(), anyBoolean()); + eh.handleOtherException(wrap(new IllegalStateException()), mock(Consumer.class), + mock(MessageListenerContainer.class), true); + verify(one).handleOtherException(any(), any(), any(), anyBoolean()); + } + + @Test + void testHandleOneDelegates() { + var def = mock(CommonErrorHandler.class); + var one = mock(CommonErrorHandler.class); + var two = mock(CommonErrorHandler.class); + var three = mock(CommonErrorHandler.class); + var eh = new CommonDelegatingErrorHandler(def); + eh.setErrorHandlers(Map.of(IllegalStateException.class, one, IllegalArgumentException.class, two)); + eh.addDelegate(RuntimeException.class, three); + + eh.handleOne(wrap(new IOException()), mock(ConsumerRecord.class), mock(Consumer.class), + mock(MessageListenerContainer.class)); + verify(def).handleOne(any(), any(), any(), any()); + eh.handleOne(wrap(new KafkaException("test")), mock(ConsumerRecord.class), mock(Consumer.class), + mock(MessageListenerContainer.class)); + verify(three).handleOne(any(), any(), any(), any()); + eh.handleOne(wrap(new IllegalArgumentException()), mock(ConsumerRecord.class), mock(Consumer.class), + mock(MessageListenerContainer.class)); + verify(two).handleOne(any(), any(), any(), any()); + eh.handleOne(wrap(new IllegalStateException()), mock(ConsumerRecord.class), mock(Consumer.class), + mock(MessageListenerContainer.class)); + verify(one).handleOne(any(), any(), any(), any()); + } + @Test void testDelegateForThrowableIsAppliedWhenCauseTraversingIsEnabled() { var defaultHandler = mock(CommonErrorHandler.class);