GH-835: LCF: check error handler is the right type

Resolves #835

Illegal state instead of `ClassCastException`.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -22,6 +22,7 @@
 import java.util.regex.Pattern;
 
 import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.kafka.core.ConsumerFactory;
@@ -38,6 +39,7 @@
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
+import org.springframework.util.Assert;
 
 /**
  * Base {@link KafkaListenerContainerFactory} for Spring's base container implementation.
@@ -53,7 +55,7 @@
  * @see AbstractMessageListenerContainer
  */
 public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMessageListenerContainer<K, V>, K, V>
-		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware {
+		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, InitializingBean {
 
 	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null);
 
@@ -251,6 +253,20 @@ public ContainerProperties getContainerProperties() {
 		return this.containerProperties;
 	}
 
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		if (this.errorHandler != null) {
+			if (Boolean.TRUE.equals(this.batchListener)) {
+				Assert.state(this.errorHandler instanceof BatchErrorHandler,
+						"The error handler must be a BatchErrorHandler, not " + this.errorHandler.getClass().getName());
+			}
+			else {
+				Assert.state(this.errorHandler instanceof ErrorHandler,
+						"The error handler must be an ErrorHandler, not " + this.errorHandler.getClass().getName());
+			}
+		}
+	}
+
 	@SuppressWarnings("unchecked")
 	@Override
 	public C createListenerContainer(KafkaListenerEndpoint endpoint) {