spring-projectsGH-2288: Add option to deeply traverse exc chain

Fixes spring-projects#2288

Add possibility to deeply traverse cause chain in order to find proper delegate
for handling thrown exception. Keep old way of cause chain traversing as default
one.

Cover new code with unit test.

Author: breader124

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

@@ -34,6 +34,7 @@
  * {@link #deliveryAttemptHeader()} is not supported - always returns false.
  *
  * @author Gary Russell
+ * @author Adrian Chlebosz
  * @since 2.8
  *
  */
@@ -43,6 +44,8 @@ public class CommonDelegatingErrorHandler implements CommonErrorHandler {
 
 	private final Map<Class<? extends Throwable>, CommonErrorHandler> delegates = new LinkedHashMap<>();
 
+	private boolean deepCauseChainTraversing = false;
+
 	/**
 	 * Construct an instance with a default error handler that will be invoked if the
 	 * exception has no matches.
@@ -64,6 +67,15 @@ public void setErrorHandlers(Map<Class<? extends Throwable>, CommonErrorHandler>
 		checkDelegates();
 	}
 
+	/**
+	 * Set the flag enabling deep exception's cause chain traversing. If true, delegate
+	 * for initial {@link Throwable} will be retrieved. Initial exception is an exception
+	 * without cause or with unknown cause.
+	 * @param deepCauseChainTraversing the deepCauseChainTraversing flag.
+	 */
+	public void setDeepCauseChainTraversing(boolean deepCauseChainTraversing) {
+		this.deepCauseChainTraversing = deepCauseChainTraversing;
+	}
 
 	@SuppressWarnings("deprecation")
 	@Override
@@ -79,7 +91,7 @@ public boolean seeksAfterHandling() {
 	@Override
 	public void clearThreadState() {
 		this.defaultErrorHandler.clearThreadState();
-		this.delegates.values().forEach(handler -> handler.clearThreadState());
+		this.delegates.values().forEach(CommonErrorHandler::clearThreadState);
 	}
 
 	@Override
@@ -158,10 +170,7 @@ public void handleOtherException(Exception thrownException, Consumer<?, ?> consu
 
 	@Nullable
 	private CommonErrorHandler findDelegate(Throwable thrownException) {
-		Throwable cause = thrownException;
-		if (cause instanceof ListenerExecutionFailedException) {
-			cause = thrownException.getCause();
-		}
+		Throwable cause = findCause(thrownException);
 		if (cause != null) {
 			Class<? extends Throwable> causeClass = cause.getClass();
 			for (Entry<Class<? extends Throwable>, CommonErrorHandler> entry : this.delegates.entrySet()) {
@@ -173,4 +182,29 @@ private CommonErrorHandler findDelegate(Throwable thrownException) {
 		return null;
 	}
 
+	@Nullable
+	private Throwable findCause(Throwable thrownException) {
+		if (this.deepCauseChainTraversing) {
+			return deepTraverseCauseChain(thrownException);
+		}
+		return shallowTraverseCauseChain(thrownException);
+	}
+
+	@Nullable
+	private Throwable shallowTraverseCauseChain(Throwable thrownException) {
+		Throwable cause = thrownException;
+		if (cause instanceof ListenerExecutionFailedException) {
+			cause = thrownException.getCause();
+		}
+		return cause;
+	}
+
+	@Nullable
+	private Throwable deepTraverseCauseChain(Throwable thrownException) {
+		while (thrownException != null && thrownException.getCause() != null) {
+			thrownException = thrownException.getCause();
+		}
+		return thrownException;
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -18,6 +18,7 @@
 
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 
 import java.io.IOException;
@@ -31,10 +32,10 @@
 import org.springframework.kafka.KafkaException;
 
 /**
- * Tests for {@link CommonDelegatingErrorHandler}. Copied from
- * {@link ConditionalDelegatingErrorHandlerTests} with changed handler type.
+ * Tests for {@link CommonDelegatingErrorHandler}.
  *
  * @author Gary Russell
+ * @author Adrian Chlebosz
  * @since 2.8
  *
  */
@@ -88,6 +89,46 @@ void testBatchDelegates() {
 		verify(one).handleBatch(any(), any(), any(), any(), any());
 	}
 
+	@Test
+	void testDelegateForInitialCauseOfThrowableIsAppliedWhenDeepTraversingFlagIsSet() {
+		var defaultHandler = mock(CommonErrorHandler.class);
+
+		var initErrorHandler = mock(CommonErrorHandler.class);
+		var initExc = new IllegalStateException();
+		var directCauseErrorHandler = mock(CommonErrorHandler.class);
+		var directCauseExc = new IllegalArgumentException(initExc);
+		var errorHandler = mock(CommonErrorHandler.class);
+		var exc = new UnsupportedOperationException(directCauseExc);
+
+		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
+		delegatingErrorHandler.setDeepCauseChainTraversing(true);
+		delegatingErrorHandler.setErrorHandlers(Map.of(
+			initExc.getClass(), initErrorHandler,
+			directCauseExc.getClass(), directCauseErrorHandler,
+			exc.getClass(), errorHandler
+		));
+
+		delegatingErrorHandler.handleRemaining(exc, Collections.emptyList(), mock(Consumer.class),
+			mock(MessageListenerContainer.class));
+
+		verify(initErrorHandler).handleRemaining(any(), any(), any(), any());
+		verify(directCauseErrorHandler, never()).handleRemaining(any(), any(), any(), any());
+		verify(errorHandler, never()).handleRemaining(any(), any(), any(), any());
+	}
+
+	@Test
+	@SuppressWarnings("ConstantConditions")
+	void testDefaultDelegateIsAppliedWhenDeepTraversingFlagIsSetAndNullThrowableHasBeenPassed() {
+		var defaultHandler = mock(CommonErrorHandler.class);
+		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
+		delegatingErrorHandler.setDeepCauseChainTraversing(true);
+
+		delegatingErrorHandler.handleRemaining(null, Collections.emptyList(), mock(Consumer.class),
+			mock(MessageListenerContainer.class));
+
+		verify(defaultHandler).handleRemaining(any(), any(), any(), any());
+	}
+
 	private Exception wrap(Exception ex) {
 		return new ListenerExecutionFailedException("test", ex);
 	}