GH-1842: Add Conditional Delegating Error Handlers

Resolves #1842

* Checkstyle fixes and polishing.

* Polishing - return after a delegate handles the error.

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -5314,6 +5314,14 @@ The `ContainerStoppingBatchErrorHandler` (used with batch listeners) stops the c
 After the container stops, an exception that wraps the `ListenerExecutionFailedException` is thrown.
 This is to cause the transaction to roll back (if transactions are enabled).
 
+[[cond-eh]]
+===== Conditional Delegating Error Handlers
+
+Introduced in version 2.7.4, the `ConditionalDelegatingErrorHandler` can delegate to different error handlers, depending on the exception type.
+For example, you may wish to invoke a `SeekToCurrentErrorHandler` for most exceptions, or a `ContainerStoppingErrorHandler` for others.
+
+Similarly, the `ConditionalDelegatingBatchErrorHandler` is provided.
+
 [[after-rollback]]
 ===== After-rollback Processor
 

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -105,3 +105,9 @@ See <<container-sequencing>> for more information.
 
 A new `BackOff` implementation is provided, making it more convenient to configure the max retries.
 See <<exp-backoff>> for more information.
+
+[[x27-delegating-eh]]
+==== Conditional Delegating Error Handlers
+
+These new error handlers can be configured to delegate to different error handlers, depending on the exception type.
+See <<cond-eh>> for more information.

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

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * An error handler that delegates to different error handlers, depending on the exception
+ * type.
+ *
+ * @author Gary Russell
+ * @since 2.7.4
+ *
+ */
+public class ConditionalDelegatingBatchErrorHandler implements ContainerAwareBatchErrorHandler {
+
+	private final ContainerAwareBatchErrorHandler defaultErrorHandler;
+
+	private final Map<Class<? extends Throwable>, ContainerAwareBatchErrorHandler> delegates = new LinkedHashMap<>();
+
+	/**
+	 * Construct an instance with a default error handler that will be invoked if the
+	 * exception has no matches.
+	 * @param defaultErrorHandler the default error handler.
+	 */
+	public ConditionalDelegatingBatchErrorHandler(ContainerAwareBatchErrorHandler defaultErrorHandler) {
+		Assert.notNull(defaultErrorHandler, "'defaultErrorHandler' cannot be null");
+		this.defaultErrorHandler = defaultErrorHandler;
+	}
+
+	/**
+	 * Set the delegate error handlers; a {@link LinkedHashMap} argument is recommended so
+	 * that the delegates are searched in a known order.
+	 * @param delegates the delegates.
+	 */
+	public void setErrorHandlers(Map<Class<? extends Throwable>, ContainerAwareBatchErrorHandler> delegates) {
+		this.delegates.clear();
+		this.delegates.putAll(delegates);
+	}
+
+	/**
+	 * Add a delegate to the end of the current collection.
+	 * @param throwable the throwable for this handler.
+	 * @param handler the handler.
+	 */
+	public void addDelegate(Class<? extends Throwable> throwable, ContainerAwareBatchErrorHandler handler) {
+		this.delegates.put(throwable, handler);
+	}
+
+	@Override
+	public void handle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		// Never called but, just in case
+		doHandle(thrownException, records, consumer, container, null);
+	}
+
+	@Override
+	public void handle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container, Runnable invokeListener) {
+
+		doHandle(thrownException, records, consumer, container, invokeListener);
+	}
+
+	protected void doHandle(Exception thrownException, ConsumerRecords<?, ?> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container, @Nullable Runnable invokeListener) {
+
+		Throwable cause = thrownException;
+		if (cause instanceof ListenerExecutionFailedException) {
+			cause = thrownException.getCause();
+		}
+		if (cause != null) {
+			Class<? extends Throwable> causeClass = cause.getClass();
+			for (Entry<Class<? extends Throwable>, ContainerAwareBatchErrorHandler> entry : this.delegates.entrySet()) {
+				if (entry.getKey().equals(causeClass)) {
+					entry.getValue().handle(thrownException, records, consumer, container, invokeListener);
+					return;
+				}
+			}
+		}
+		this.defaultErrorHandler.handle(thrownException, records, consumer, container, invokeListener);
+	}
+
+}

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

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * An error handler that delegates to different error handlers, depending on the exception
+ * type.
+ *
+ * @author Gary Russell
+ * @since 2.7.4
+ *
+ */
+public class ConditionalDelegatingErrorHandler implements ContainerAwareErrorHandler {
+
+	private final ContainerAwareErrorHandler defaultErrorHandler;
+
+	private final Map<Class<? extends Throwable>, ContainerAwareErrorHandler> delegates = new LinkedHashMap<>();
+
+	/**
+	 * Construct an instance with a default error handler that will be invoked if the
+	 * exception has no matches.
+	 * @param defaultErrorHandler the default error handler.
+	 */
+	public ConditionalDelegatingErrorHandler(ContainerAwareErrorHandler defaultErrorHandler) {
+		Assert.notNull(defaultErrorHandler, "'defaultErrorHandler' cannot be null");
+		this.defaultErrorHandler = defaultErrorHandler;
+	}
+
+	/**
+	 * Set the delegate error handlers; a {@link LinkedHashMap} argument is recommended so
+	 * that the delegates are searched in a known order.
+	 * @param delegates the delegates.
+	 */
+	public void setErrorHandlers(Map<Class<? extends Throwable>, ContainerAwareErrorHandler> delegates) {
+		this.delegates.clear();
+		this.delegates.putAll(delegates);
+	}
+
+	/**
+	 * Add a delegate to the end of the current collection.
+	 * @param throwable the throwable for this handler.
+	 * @param handler the handler.
+	 */
+	public void addDelegate(Class<? extends Throwable> throwable, ContainerAwareErrorHandler handler) {
+		this.delegates.put(throwable, handler);
+	}
+
+	@Override
+	public void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		boolean handled = false;
+		Throwable cause = thrownException;
+		if (cause instanceof ListenerExecutionFailedException) {
+			cause = thrownException.getCause();
+		}
+		if (cause != null) {
+			Class<? extends Throwable> causeClass = cause.getClass();
+			for (Entry<Class<? extends Throwable>, ContainerAwareErrorHandler> entry : this.delegates.entrySet()) {
+				if (entry.getKey().equals(causeClass)) {
+					handled = true;
+					entry.getValue().handle(thrownException, records, consumer, container);
+					return;
+				}
+			}
+		}
+		this.defaultErrorHandler.handle(thrownException, records, consumer, container);
+	}
+
+}

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

@@ -21,6 +21,8 @@
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
+import org.springframework.lang.Nullable;
+
 /**
  * An error handler that has access to the unprocessed records from the last poll
  * (including the failed record), the consumer, and the container.
@@ -35,12 +37,14 @@
 public interface ContainerAwareErrorHandler extends RemainingRecordsErrorHandler {
 
 	@Override
-	default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer) {
+	default void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records,
+			Consumer<?, ?> consumer) {
+
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
 	@Override
-	void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
+	void handle(Exception thrownException, @Nullable List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container);
 
 }

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2021 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.junit.jupiter.api.Test;
+
+/**
+ * @author Gary Russell
+ * @since 2.7.4
+ *
+ */
+public class ConditionalDelegatingErrorHandlerTests {
+
+	@Test
+	void testRecordDelegates() {
+		var def = mock(ContainerAwareErrorHandler.class);
+		var one = mock(ContainerAwareErrorHandler.class);
+		var two = mock(ContainerAwareErrorHandler.class);
+		var three = mock(ContainerAwareErrorHandler.class);
+		var eh = new ConditionalDelegatingErrorHandler(def);
+		eh.setErrorHandlers(Map.of(IllegalStateException.class, one, IllegalArgumentException.class, two));
+		eh.addDelegate(RuntimeException.class, three);
+
+		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),
+				mock(MessageListenerContainer.class));
+		verify(three).handle(any(), any(), any(), any());
+		eh.handle(wrap(new IllegalArgumentException()), Collections.emptyList(), mock(Consumer.class),
+				mock(MessageListenerContainer.class));
+		verify(two).handle(any(), any(), any(), any());
+		eh.handle(wrap(new IllegalStateException()), Collections.emptyList(), mock(Consumer.class),
+				mock(MessageListenerContainer.class));
+		verify(one).handle(any(), any(), any(), any());
+	}
+
+	@Test
+	void testBatchDelegates() {
+		var def = mock(ContainerAwareBatchErrorHandler.class);
+		var one = mock(ContainerAwareBatchErrorHandler.class);
+		var two = mock(ContainerAwareBatchErrorHandler.class);
+		var three = mock(ContainerAwareBatchErrorHandler.class);
+		var eh = new ConditionalDelegatingBatchErrorHandler(def);
+		eh.setErrorHandlers(Map.of(IllegalStateException.class, one, IllegalArgumentException.class, two));
+		eh.addDelegate(RuntimeException.class, three);
+
+		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),
+				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),
+				mock(MessageListenerContainer.class), mock(Runnable.class));
+		verify(two).handle(any(), any(), any(), any(), any());
+		eh.handle(wrap(new IllegalStateException()), mock(ConsumerRecords.class), mock(Consumer.class),
+				mock(MessageListenerContainer.class), mock(Runnable.class));
+		verify(one).handle(any(), any(), any(), any(), any());
+	}
+
+	private Exception wrap(Exception ex) {
+		return new ListenerExecutionFailedException("test", ex);
+	}
+
+}