Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3227: Implement handleOne() in CDEH

Browse Source 
Fixes: #3227 

* Implement handleOne() in `CommonDelegatingErrorHandler`
* Add tests for handle methods in `CommonDelegatingErrorHandler`
* Add tests for the new `handleOne()` method, as well as a test for `handleOtherException()`
* Checkstyle fixes

**Auto-cherry-pick to `3.1.x` & `3.0.x`**
This commit is contained in:
Dan Blackney
2024-05-03 22:28:22 +08:00
committed by GitHub
parent b05510df77
commit 4e06c2c165
2 changed files with 67 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-kafka/src/main/java/org/springframework/kafka/listener/CommonDelegatingErrorHandler.java
View File

@@ -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);

55
spring-kafka/src/test/java/org/springframework/kafka/listener/CommonDelegatingErrorHandlerTests.java
View File

@@ -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);