GH-926: Fix error handler detection

Fixes #926

- Use correct wired error handler for key/value
- Fix sonar issue about the properties being nullable

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/ConsumerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -103,6 +103,7 @@ default Map<String, Object> getConfigurationProperties() {
 	 * @return the deserializer.
 	 * @since 2.0
 	 */
+	@Nullable
 	default Deserializer<K> getKeyDeserializer() {
 		throw new UnsupportedOperationException("'getKeyDeserializer()' is not supported");
 	}
@@ -113,6 +114,7 @@ default Deserializer<K> getKeyDeserializer() {
 	 * @return the deserializer.
 	 * @since 2.0
 	 */
+	@Nullable
 	default Deserializer<V> getValueDeserializer() {
 		throw new UnsupportedOperationException("'getKeyDeserializer()' is not supported");
 	}

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

@@ -546,16 +546,18 @@ else if (listener instanceof MessageListener) {
 				this.logger.info(this);
 			}
 			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
-			this.checkNullKeyForExceptions = checkDeserializer(
-					findDeserializerClass(props, ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
-			this.checkNullValueForExceptions = checkDeserializer(
-					findDeserializerClass(props, ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
+			this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, false));
+			this.checkNullValueForExceptions = checkDeserializer(findDeserializerClass(props, true));
 		}
 
-		private Object findDeserializerClass(Map<String, Object> props, String config) {
-			Object configuredDeserializer = KafkaMessageListenerContainer.this.consumerFactory.getKeyDeserializer();
+		private Object findDeserializerClass(Map<String, Object> props, boolean isValue) {
+			Object configuredDeserializer = isValue
+					? KafkaMessageListenerContainer.this.consumerFactory.getValueDeserializer()
+					: KafkaMessageListenerContainer.this.consumerFactory.getKeyDeserializer();
 			if (configuredDeserializer == null) {
-				return props.get(config);
+				return props.get(isValue
+						? ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
+						: ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
 			}
 			else {
 				return configuredDeserializer.getClass();