spring-projectsGH-2288: Make cause traversing more extensible

Use BinaryExceptionClassifier while traversing cause chain to make it possible
to classify throwables for handling based on inheritance etc.

Author: breader124

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

@@ -16,15 +16,18 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.stream.Collectors;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
+import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -44,7 +47,8 @@ public class CommonDelegatingErrorHandler implements CommonErrorHandler {
 
 	private final Map<Class<? extends Throwable>, CommonErrorHandler> delegates = new LinkedHashMap<>();
 
-	private boolean deepCauseChainTraversing = false;
+	private boolean causeChainTraversing = false;
+	private NoTraversingBinaryExceptionClassifier classifier = new NoTraversingBinaryExceptionClassifier(new HashMap<>());
 
 	/**
 	 * Construct an instance with a default error handler that will be invoked if the
@@ -62,19 +66,28 @@ public CommonDelegatingErrorHandler(CommonErrorHandler defaultErrorHandler) {
 	 * @param delegates the delegates.
 	 */
 	public void setErrorHandlers(Map<Class<? extends Throwable>, CommonErrorHandler> delegates) {
+		Assert.notNull(delegates, "'delegates' cannot be null");
 		this.delegates.clear();
 		this.delegates.putAll(delegates);
 		checkDelegates();
+		updateClassifier(delegates);
+	}
+
+	private void updateClassifier(Map<Class<? extends Throwable>, CommonErrorHandler> delegates) {
+		Map<Class<? extends Throwable>, Boolean> classifications = delegates.keySet().stream()
+			.map(commonErrorHandler -> Map.entry(commonErrorHandler, true))
+			.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+		this.classifier = new NoTraversingBinaryExceptionClassifier(classifications);
 	}
 
 	/**
 	 * 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.
+	 * @param causeChainTraversing the deepCauseChainTraversing flag.
 	 */
-	public void setDeepCauseChainTraversing(boolean deepCauseChainTraversing) {
-		this.deepCauseChainTraversing = deepCauseChainTraversing;
+	public void setCauseChainTraversing(boolean causeChainTraversing) {
+		this.causeChainTraversing = causeChainTraversing;
 	}
 
 	@SuppressWarnings("deprecation")
@@ -184,8 +197,8 @@ private CommonErrorHandler findDelegate(Throwable thrownException) {
 
 	@Nullable
 	private Throwable findCause(Throwable thrownException) {
-		if (this.deepCauseChainTraversing) {
-			return deepTraverseCauseChain(thrownException);
+		if (this.causeChainTraversing) {
+			return traverseCauseChain(thrownException);
 		}
 		return shallowTraverseCauseChain(thrownException);
 	}
@@ -200,11 +213,23 @@ private Throwable shallowTraverseCauseChain(Throwable thrownException) {
 	}
 
 	@Nullable
-	private Throwable deepTraverseCauseChain(Throwable thrownException) {
+	private Throwable traverseCauseChain(Throwable thrownException) {
 		while (thrownException != null && thrownException.getCause() != null) {
+			if (this.classifier.classify(thrownException)) { // NOSONAR using Boolean here is not dangerous
+				return thrownException;
+			}
 			thrownException = thrownException.getCause();
 		}
 		return thrownException;
 	}
 
+	@SuppressWarnings("serial")
+	private static final class NoTraversingBinaryExceptionClassifier extends BinaryExceptionClassifier {
+
+		NoTraversingBinaryExceptionClassifier(Map<Class<? extends Throwable>, Boolean> typeMap) {
+			super(typeMap, false, false);
+		}
+
+	}
+
 }

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

@@ -30,6 +30,7 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.core.KafkaProducerException;
 
 /**
  * Tests for {@link CommonDelegatingErrorHandler}.
@@ -90,38 +91,75 @@ void testBatchDelegates() {
 	}
 
 	@Test
-	void testDelegateForInitialCauseOfThrowableIsAppliedWhenDeepTraversingFlagIsSet() {
+	void testDelegateForThrowableIsAppliedWhenCauseTraversingIsEnabled() {
 		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 directCauseExc = new IllegalArgumentException();
 		var errorHandler = mock(CommonErrorHandler.class);
 		var exc = new UnsupportedOperationException(directCauseExc);
 
 		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
-		delegatingErrorHandler.setDeepCauseChainTraversing(true);
+		delegatingErrorHandler.setCauseChainTraversing(true);
 		delegatingErrorHandler.setErrorHandlers(Map.of(
-			initExc.getClass(), initErrorHandler,
-			directCauseExc.getClass(), directCauseErrorHandler,
-			exc.getClass(), errorHandler
+			exc.getClass(), errorHandler,
+			directCauseExc.getClass(), directCauseErrorHandler
 		));
 
-		delegatingErrorHandler.handleRemaining(exc, Collections.emptyList(), mock(Consumer.class),
+		delegatingErrorHandler.handleRemaining(directCauseExc, Collections.emptyList(), mock(Consumer.class),
 			mock(MessageListenerContainer.class));
 
-		verify(initErrorHandler).handleRemaining(any(), any(), any(), any());
-		verify(directCauseErrorHandler, never()).handleRemaining(any(), any(), any(), any());
+		verify(directCauseErrorHandler).handleRemaining(any(), any(), any(), any());
 		verify(errorHandler, never()).handleRemaining(any(), any(), any(), any());
 	}
 
+	@Test
+	void testDelegateForThrowableCauseIsAppliedWhenCauseTraversingIsEnabled() {
+		var defaultHandler = mock(CommonErrorHandler.class);
+
+		var directCauseErrorHandler = mock(CommonErrorHandler.class);
+		var directCauseExc = new IllegalArgumentException();
+		var exc = new UnsupportedOperationException(directCauseExc);
+
+		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
+		delegatingErrorHandler.setCauseChainTraversing(true);
+		delegatingErrorHandler.setErrorHandlers(Map.of(
+			directCauseExc.getClass(), directCauseErrorHandler
+		));
+
+		delegatingErrorHandler.handleRemaining(exc, Collections.emptyList(), mock(Consumer.class),
+			mock(MessageListenerContainer.class));
+
+		verify(directCauseErrorHandler).handleRemaining(any(), any(), any(), any());
+	}
+
+	@Test
+	@SuppressWarnings("ConstantConditions")
+	void testDelegateForClassifiableThrowableCauseIsAppliedWhenCauseTraversingIsEnabled() {
+		var defaultHandler = mock(CommonErrorHandler.class);
+
+		var directCauseErrorHandler = mock(CommonErrorHandler.class);
+		var directCauseExc = new KafkaProducerException(null, null, null);
+		var exc = new UnsupportedOperationException(directCauseExc);
+
+		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
+		delegatingErrorHandler.setCauseChainTraversing(true);
+		delegatingErrorHandler.setErrorHandlers(Map.of(
+			KafkaException.class, directCauseErrorHandler
+		));
+
+		delegatingErrorHandler.handleRemaining(exc, Collections.emptyList(), mock(Consumer.class),
+			mock(MessageListenerContainer.class));
+
+		verify(directCauseErrorHandler).handleRemaining(any(), any(), any(), any());
+	}
+
 	@Test
 	@SuppressWarnings("ConstantConditions")
-	void testDefaultDelegateIsAppliedWhenDeepTraversingFlagIsSetAndNullThrowableHasBeenPassed() {
+	void testDefaultDelegateIsApplied() {
 		var defaultHandler = mock(CommonErrorHandler.class);
 		var delegatingErrorHandler = new CommonDelegatingErrorHandler(defaultHandler);
-		delegatingErrorHandler.setDeepCauseChainTraversing(true);
+		delegatingErrorHandler.setCauseChainTraversing(true);
 
 		delegatingErrorHandler.handleRemaining(null, Collections.emptyList(), mock(Consumer.class),
 			mock(MessageListenerContainer.class));