GH-926: Always Detect ErrorHandlingDeserializer2

Fixes #926

Previously, the container only detected the presence of the deserializer
when it was configured using Kafka properties.

The container now detects it (and subclasses) when configured directly
into the consumer factory and properly routes exceptions to the error
handler.

Also add `setKey` to designate the deserializer as a key deserializer.

* Polishing - PR Comments.

* Polishing

Author: garyrussell

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

@@ -546,9 +546,20 @@ else if (listener instanceof MessageListener) {
 				this.logger.info(this);
 			}
 			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
-			this.checkNullKeyForExceptions = checkDeserializer(props.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
+			this.checkNullKeyForExceptions = checkDeserializer(
+					findDeserializerClass(props, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
 			this.checkNullValueForExceptions = checkDeserializer(
-					props.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
+					findDeserializerClass(props, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
+		}
+
+		private Object findDeserializerClass(Map<String, Object> props, String config) {
+			Object configuredDeserializer = KafkaMessageListenerContainer.this.consumerFactory.getKeyDeserializer();
+			if (configuredDeserializer == null) {
+				return props.get(config);
+			}
+			else {
+				return configuredDeserializer.getClass();
+			}
 		}
 
 		private void subscribeOrAssignTopics(final Consumer<K, V> consumer) {
@@ -576,7 +587,7 @@ private void subscribeOrAssignTopics(final Consumer<K, V> consumer) {
 
 		private boolean checkDeserializer(Object deser) {
 			return deser instanceof Class
-					? ((Class<?>) deser).equals(ErrorHandlingDeserializer2.class)
+					? ErrorHandlingDeserializer2.class.isAssignableFrom((Class<?>) deser)
 					: deser instanceof String
 						? ((String) deser).equals(ErrorHandlingDeserializer2.class.getName())
 						: false;

spring-kafka/src/main/java/org/springframework/kafka/support/serializer/ErrorHandlingDeserializer2.java

@@ -97,6 +97,31 @@ public void setFailedDeserializationFunction(BiFunction<byte[], Headers, T> fail
 		this.failedDeserializationFunction = failedDeserializationFunction;
 	}
 
+	public boolean isKey() {
+		return this.isKey;
+	}
+
+	/**
+	 * Set to true if this deserializer is to be used as a key deserializer when
+	 * configuring outside of Kafka.
+	 * @param isKey true for a key deserializer, false otherwise.
+	 * @since 2.2.3
+	 */
+	public void setKey(boolean isKey) {
+		this.isKey = isKey;
+	}
+
+	/**
+	 * Set to true if this deserializer is to be used as a key deserializer when
+	 * configuring outside of Kafka.
+	 * @param isKey true for a key deserializer, false otherwise.
+	 * @return this
+	 * @since 2.2.3
+	 */
+	public ErrorHandlingDeserializer2<T> keyDeserializer(boolean isKey) {
+		this.isKey = isKey;
+		return this;
+	}
 
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {

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

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -74,9 +75,9 @@ public void testBadDeserializer() throws Exception {
 		this.config.template().send(TOPIC, "fail", "bar");
 		this.config.template().send(TOPIC, "foo", "fail");
 		assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
-		assertThat(this.config.goodCount).isEqualTo(1);
-		assertThat(this.config.keyErrorCount).isEqualTo(1);
-		assertThat(this.config.valueErrorCount).isEqualTo(1);
+		assertThat(this.config.goodCount.get()).withFailMessage("Counts wrong: %s", this.config).isEqualTo(2);
+		assertThat(this.config.keyErrorCount.get()).withFailMessage("Counts wrong: %s", this.config).isEqualTo(2);
+		assertThat(this.config.valueErrorCount.get()).withFailMessage("Counts wrong: %s", this.config).isEqualTo(2);
 		assertThat(this.config.headers).isNotNull();
 	}
 
@@ -113,19 +114,25 @@ public void close() {
 	@EnableKafka
 	public static class Config {
 
-		private final CountDownLatch latch = new CountDownLatch(3);
+		private final CountDownLatch latch = new CountDownLatch(6);
 
-		private int goodCount;
+		private final AtomicInteger goodCount = new AtomicInteger();
 
-		private int keyErrorCount;
+		private final AtomicInteger keyErrorCount = new AtomicInteger();
 
-		private int valueErrorCount;
+		private final AtomicInteger valueErrorCount = new AtomicInteger();
 
 		private Headers headers;
 
 		@KafkaListener(topics = TOPIC)
-		public void listen(ConsumerRecord<String, String> record) {
-			this.goodCount++;
+		public void listen1(ConsumerRecord<String, String> record) {
+			this.goodCount.incrementAndGet();
+			this.latch.countDown();
+		}
+
+		@KafkaListener(topics = TOPIC, containerFactory = "kafkaListenerContainerFactoryExplicitDesers")
+		public void listen2(ConsumerRecord<String, String> record) {
+			this.goodCount.incrementAndGet();
 			this.latch.countDown();
 		}
 
@@ -136,16 +143,25 @@ public EmbeddedKafkaBroker embeddedKafka() {
 
 		@Bean
 		public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
+			return factory(cf());
+		}
+
+		@Bean
+		public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactoryExplicitDesers() {
+			return factory(cfWithExplicitDeserializers());
+		}
+
+		private ConcurrentKafkaListenerContainerFactory<String, String> factory(ConsumerFactory<String, String> cf) {
 			ConcurrentKafkaListenerContainerFactory<String, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
-			factory.setConsumerFactory(cf());
+			factory.setConsumerFactory(cf);
 			factory.setErrorHandler((t, r) -> {
 				if (r.value() == null && t instanceof DeserializationException) {
-					this.valueErrorCount++;
+					this.valueErrorCount.incrementAndGet();
 					this.headers = ((DeserializationException) t).getHeaders();
 				}
 				else if (r.key() == null && t instanceof DeserializationException) {
-					this.keyErrorCount++;
+					this.keyErrorCount.incrementAndGet();
 				}
 				this.latch.countDown();
 			});
@@ -154,7 +170,7 @@ else if (r.key() == null && t instanceof DeserializationException) {
 
 		@Bean
 		public ConsumerFactory<String, String> cf() {
-			Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC, "false", embeddedKafka());
+			Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC + ".g1", "false", embeddedKafka());
 			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
 			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer2.class);
@@ -163,6 +179,15 @@ public ConsumerFactory<String, String> cf() {
 			return new DefaultKafkaConsumerFactory<>(props);
 		}
 
+		@Bean
+		public ConsumerFactory<String, String> cfWithExplicitDeserializers() {
+			Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC + ".g2", "false", embeddedKafka());
+			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+			return new DefaultKafkaConsumerFactory<>(props,
+					new ErrorHandlingDeserializer2<String>(new FailSometimesDeserializer()).keyDeserializer(true),
+					new ErrorHandlingDeserializer2<String>(new FailSometimesDeserializer()));
+		}
+
 		@Bean
 		public ProducerFactory<String, String> pf() {
 			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka());
@@ -175,6 +200,12 @@ public KafkaTemplate<String, String> template() {
 			return new KafkaTemplate<>(pf());
 		}
 
+		@Override
+		public String toString() {
+			return "Config [goodCount=" + this.goodCount.get() + ", keyErrorCount=" + this.keyErrorCount.get()
+					+ ", valueErrorCount=" + this.valueErrorCount.get() + "]";
+		}
+
 	}
 
 	public static class FailSometimesDeserializer implements ExtendedDeserializer<String> {