GH-2050: Configure ErrHandlDeser if no delegate

Fixes #2050

The application can have several consumer factories when one fully relies
on the configuration properties for its deserializers and other
configures them programmatically.
The consumer factory now calls `configure()` on the `Deserializer`
independently of its origins.
See #1879
In this case the `ErrorHandlingDeserializer` consults
`spring.deserializer.key.delegate.class` or `spring.deserializer.value.delegate.class`
for its delegate overriding provided explicitly programmatically

* Fix `ErrorHandlingDeserializer` to check it `delegate` and `failedDeserializationFunction`
for null before taking their values from the respective configuration properties
* Add `spring.deserializer.value.delegate.class` property to `testJsonSerDeIgnoreTypeHeadersInbound()`
configuration to ensure that it does not override an explicit `JsonDeserializer` delegate
* Fix warning in the `EmbeddedKafkaBroker` about `this.` prefix for a `static logger` property

Author: artembilan

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -595,7 +595,7 @@ private boolean brokerRunning(KafkaServer kafkaServer) {
 				}
 			}
 			else {
-				this.logger.debug("Could not determine broker state during shutdown");
+				logger.debug("Could not determine broker state during shutdown");
 				return true;
 			}
 		}

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

@@ -132,11 +132,15 @@ public ErrorHandlingDeserializer<T> keyDeserializer(boolean isKey) {
 
 	@Override
 	public void configure(Map<String, ?> configs, boolean isKey) {
-		setupDelegate(configs, isKey ? KEY_DESERIALIZER_CLASS : VALUE_DESERIALIZER_CLASS);
+		if (this.delegate == null) {
+			setupDelegate(configs, isKey ? KEY_DESERIALIZER_CLASS : VALUE_DESERIALIZER_CLASS);
+		}
 		Assert.state(this.delegate != null, "No delegate deserializer configured");
 		this.delegate.configure(configs, isKey);
 		this.isForKey = isKey;
-		setupFunction(configs, isKey ? KEY_FUNCTION : VALUE_FUNCTION);
+		if (this.failedDeserializationFunction == null) {
+			setupFunction(configs, isKey ? KEY_FUNCTION : VALUE_FUNCTION);
+		}
 	}
 
 	public void setupDelegate(Map<String, ?> configs, String configKey) {

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

@@ -2257,7 +2257,8 @@ public void testJsonSerDeTypeMappings() throws Exception {
 	public void testJsonSerDeIgnoreTypeHeadersInbound() throws Exception {
 		this.logger.info("Start JSON4");
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
-
+		props.put("spring.deserializer.value.delegate.class",
+				"org.apache.kafka.common.serialization.StringDeserializer");
 		ErrorHandlingDeserializer<Foo1> errorHandlingDeserializer =
 				new ErrorHandlingDeserializer<>(new JsonDeserializer<>(Foo1.class, false));