GH-1843: Delegating EH - use isAssignable()

Resolves #1842

Author: garyrussell

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

@@ -94,7 +94,7 @@ protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> records
 		if (cause != null) {
 			Class<? extends Throwable> causeClass = cause.getClass();
 			for (Entry<Class<? extends Throwable>, ContainerAwareBatchErrorHandler> entry : this.delegates.entrySet()) {
-				if (entry.getKey().equals(causeClass)) {
+				if (entry.getKey().isAssignableFrom(causeClass)) {
 					entry.getValue().handle(thrownException, records, consumer, container, invokeListener);
 					return;
 				}

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

@@ -82,7 +82,7 @@ public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?
 		if (cause != null) {
 			Class<? extends Throwable> causeClass = cause.getClass();
 			for (Entry<Class<? extends Throwable>, ContainerAwareErrorHandler> entry : this.delegates.entrySet()) {
-				if (entry.getKey().equals(causeClass)) {
+				if (entry.getKey().isAssignableFrom(causeClass)) {
 					handled = true;
 					entry.getValue().handle(thrownException, records, consumer, container);
 					return;

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

@@ -28,6 +28,8 @@
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.kafka.KafkaException;
+
 /**
  * @author Gary Russell
  * @since 2.7.4
@@ -48,7 +50,7 @@ void testRecordDelegates() {
 		eh.handle(wrap(new IOException()), Collections.emptyList(), mock(Consumer.class),
 				mock(MessageListenerContainer.class));
 		verify(def).handle(any(), any(), any(), any());
-		eh.handle(wrap(new RuntimeException()), Collections.emptyList(), mock(Consumer.class),
+		eh.handle(wrap(new KafkaException("test")), Collections.emptyList(), mock(Consumer.class),
 				mock(MessageListenerContainer.class));
 		verify(three).handle(any(), any(), any(), any());
 		eh.handle(wrap(new IllegalArgumentException()), Collections.emptyList(), mock(Consumer.class),
@@ -72,7 +74,7 @@ void testBatchDelegates() {
 		eh.handle(wrap(new IOException()), mock(ConsumerRecords.class), mock(Consumer.class),
 				mock(MessageListenerContainer.class), mock(Runnable.class));
 		verify(def).handle(any(), any(), any(), any(), any());
-		eh.handle(wrap(new RuntimeException()), mock(ConsumerRecords.class), mock(Consumer.class),
+		eh.handle(wrap(new KafkaException("test")), mock(ConsumerRecords.class), mock(Consumer.class),
 				mock(MessageListenerContainer.class), mock(Runnable.class));
 		verify(three).handle(any(), any(), any(), any(), any());
 		eh.handle(wrap(new IllegalArgumentException()), mock(ConsumerRecords.class), mock(Consumer.class),