GH-2046: Add Property to Always Check for Deser Ex

Resolves #2046

* Fix javadoc copy/paste.

Author: garyrussell

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

@@ -2443,6 +2443,16 @@ When null, such exceptions are considered fatal and the container will stop.
 |A prefix for the `client.id` consumer property.
 Overrides the consumer factory `client.id` property; in a concurrent container, `-n` is added as a suffix for each consumer instance.
 
+|[[checkDeserExWhenKeyNull]]<<checkDeserExWhenKeyNull,`checkDeserExWhenKeyNull`>>
+|false
+|Set to `true` to always check for a `DeserializationException` header when a `null` `key` is received.
+Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer.
+
+|[[checkDeserExWhenValueNull]]<<checkDeserExWhenValueNull,`checkDeserExWhenValueNull`>>
+|false
+|Set to `true` to always check for a `DeserializationException` header when a `null` `value` is received.
+Useful when the consumer code cannot determine that an `ErrorHandlingDeserializer` has been configured, such as when using a delegating deserializer.
+
 |[[commitCallback]]<<commitCallback,`commitCallback`>>
 |`null`
 |When present and `syncCommits` is `false` a callback invoked after the commit completes.

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

@@ -107,6 +107,10 @@ public class ConsumerProperties {
 
 	private boolean fixTxOffsets;
 
+	private boolean checkDeserExWhenKeyNull;
+
+	private boolean checkDeserExWhenValueNull;
+
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
 	 * @param topics the topics.
@@ -466,6 +470,52 @@ public void setFixTxOffsets(boolean fixTxOffsets) {
 		this.fixTxOffsets = fixTxOffsets;
 	}
 
+	/**
+	 * Always check for a deserialization exception header with a null key.
+	 * @return true to check.
+	 * @since 2.8.1
+	 */
+	public boolean isCheckDeserExWhenKeyNull() {
+		return this.checkDeserExWhenKeyNull;
+	}
+
+	/**
+	 * Set to true to always check for
+	 * {@link org.springframework.kafka.support.serializer.DeserializationException}
+	 * header when a null key is received. Useful when the consumer code cannot determine
+	 * that an
+	 * {@link org.springframework.kafka.support.serializer.ErrorHandlingDeserializer} has
+	 * been configured, such as when using a delegating deserializer.
+	 * @param checkDeserExWhenKeyNull true to always check.
+	 * @since 2.8.1
+	 */
+	public void setCheckDeserExWhenKeyNull(boolean checkDeserExWhenKeyNull) {
+		this.checkDeserExWhenKeyNull = checkDeserExWhenKeyNull;
+	}
+
+	/**
+	 * Always check for a deserialization exception header with a null value.
+	 * @return true to check.
+	 * @since 2.8.1
+	 */
+	public boolean isCheckDeserExWhenValueNull() {
+		return this.checkDeserExWhenValueNull;
+	}
+
+	/**
+	 * Set to true to always check for
+	 * {@link org.springframework.kafka.support.serializer.DeserializationException}
+	 * header when a null value is received. Useful when the consumer code cannot
+	 * determine that an
+	 * {@link org.springframework.kafka.support.serializer.ErrorHandlingDeserializer} has
+	 * been configured, such as when using a delegating deserializer.
+	 * @param checkDeserExWhenValueNull true to always check.
+	 * @since 2.8.1
+	 */
+	public void setCheckDeserExWhenValueNull(boolean checkDeserExWhenValueNull) {
+		this.checkDeserExWhenValueNull = checkDeserExWhenValueNull;
+	}
+
 	@Override
 	public String toString() {
 		return "ConsumerProperties ["

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

@@ -810,8 +810,10 @@ else if (listener instanceof MessageListener) {
 				this.logger.info(toString());
 			}
 			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
-			this.checkNullKeyForExceptions = checkDeserializer(findDeserializerClass(props, consumerProperties, false));
-			this.checkNullValueForExceptions = checkDeserializer(findDeserializerClass(props, consumerProperties, true));
+			this.checkNullKeyForExceptions = this.containerProperties.isCheckDeserExWhenKeyNull()
+					|| checkDeserializer(findDeserializerClass(props, consumerProperties, false));
+			this.checkNullValueForExceptions = this.containerProperties.isCheckDeserExWhenValueNull()
+					|| checkDeserializer(findDeserializerClass(props, consumerProperties, true));
 			this.syncCommitTimeout = determineSyncCommitTimeout();
 			if (this.containerProperties.getSyncCommitTimeout() == null) {
 				// update the property so we can use it directly from code elsewhere

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/SerializationIntegrationTests.java

@@ -58,6 +58,8 @@ void configurePreLoadedDelegates() {
 				new DefaultKafkaConsumerFactory<String, Object>(consumerProps,
 					new StringDeserializer(), delegating);
 		ContainerProperties props = new ContainerProperties(DBTD_TOPIC);
+		props.setCheckDeserExWhenKeyNull(true);
+		props.setCheckDeserExWhenValueNull(true);
 		props.setMessageListener(mock(MessageListener.class));
 		KafkaMessageListenerContainer<String, Object> container = new KafkaMessageListenerContainer<>(cFact, props);
 		container.start();
@@ -67,6 +69,10 @@ void configurePreLoadedDelegates() {
 		assertThat(delegates).hasSize(1);
 		assertThat(delegates.values().iterator().next()).isSameAs(testDeser);
 		assertThat(testDeser.configured).isTrue();
+		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.checkNullKeyForExceptions",
+				Boolean.class)).isTrue();
+		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.checkNullValueForExceptions",
+				Boolean.class)).isTrue();
 		container.stop();
 	}