Refactor DeserializationException Detection Code

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -82,6 +82,7 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass())); // NOSONAR
 
+	@NonNull
 	protected final ConsumerFactory<K, V> consumerFactory; // NOSONAR (final)
 
 	private final ContainerProperties containerProperties;

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

@@ -18,10 +18,13 @@
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.function.BiConsumer;
 
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
@@ -30,7 +33,11 @@
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.support.KafkaUtils;
+import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.backoff.BackOff;
 import org.springframework.util.backoff.BackOffExecution;
 
@@ -256,4 +263,63 @@ public static Exception findRootCause(Exception exception) {
 		return realException;
 	}
 
+	/**
+	 * Determine whether the key or value deserializer is an instance of
+	 * {@link ErrorHandlingDeserializer}.
+	 * @param <K> the key type.
+	 * @param <V> the value type.
+	 * @param consumerFactory the consumer factory.
+	 * @param consumerOverrides consumer factory property overrides.
+	 * @param isValue true to find the value deserializer.
+	 * @param classLoader the class loader to load the deserializer class.
+	 * @return true if the deserializer is an instance of
+	 * {@link ErrorHandlingDeserializer}.
+	 * @since 3.0.10
+	 */
+	public static <K, V> boolean checkDeserializer(ConsumerFactory<K, V> consumerFactory,
+			Properties consumerOverrides, boolean isValue, ClassLoader classLoader) {
+
+		Object deser = findDeserializerClass(consumerFactory, consumerOverrides, isValue);
+		Class<?> deserializer = null;
+		if (deser instanceof Class<?> deserClass) {
+			deserializer = deserClass;
+		}
+		else if (deser instanceof String str) {
+			try {
+				deserializer = ClassUtils.forName(str, classLoader);
+			}
+			catch (ClassNotFoundException | LinkageError e) {
+				throw new IllegalStateException(e);
+			}
+		}
+		else if (deser != null) {
+			throw new IllegalStateException("Deserializer must be a class or class name, not a " + deser.getClass());
+		}
+		return deserializer != null && ErrorHandlingDeserializer.class.isAssignableFrom(deserializer);
+	}
+
+	@Nullable
+	private static <K, V> Object findDeserializerClass(ConsumerFactory<K, V> consumerFactory,
+			Properties consumerOverrides, boolean isValue) {
+
+		Map<String, Object> props = consumerFactory.getConfigurationProperties();
+		Object configuredDeserializer = isValue
+				? consumerFactory.getValueDeserializer()
+				: consumerFactory.getKeyDeserializer();
+		if (configuredDeserializer == null) {
+			Object deser = consumerOverrides.get(isValue
+					? ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
+					: ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
+			if (deser == null) {
+				deser = props.get(isValue
+						? ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
+						: ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
+			}
+			return deser;
+		}
+		else {
+			return configuredDeserializer.getClass();
+		}
+	}
+
 }

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

@@ -115,7 +115,6 @@
 import org.springframework.kafka.support.micrometer.KafkaRecordReceiverContext;
 import org.springframework.kafka.support.micrometer.MicrometerHolder;
 import org.springframework.kafka.support.serializer.DeserializationException;
-import org.springframework.kafka.support.serializer.ErrorHandlingDeserializer;
 import org.springframework.kafka.support.serializer.SerializationUtils;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.lang.Nullable;
@@ -129,7 +128,6 @@
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.transaction.support.TransactionTemplate;
 import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
@@ -919,10 +917,19 @@ else if (listener instanceof MessageListener) {
 				this.logger.info(toString());
 			}
 			Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
+			ApplicationContext applicationContext = getApplicationContext();
 			this.checkNullKeyForExceptions = this.containerProperties.isCheckDeserExWhenKeyNull()
-					|| checkDeserializer(findDeserializerClass(props, consumerProperties, false));
+					|| ErrorHandlingUtils.checkDeserializer(KafkaMessageListenerContainer.this.consumerFactory,
+							consumerProperties, false,
+							applicationContext == null
+									? getClass().getClassLoader()
+									: applicationContext.getClassLoader());
 			this.checkNullValueForExceptions = this.containerProperties.isCheckDeserExWhenValueNull()
-					|| checkDeserializer(findDeserializerClass(props, consumerProperties, true));
+					|| ErrorHandlingUtils.checkDeserializer(KafkaMessageListenerContainer.this.consumerFactory,
+							consumerProperties, true,
+							applicationContext == null
+									? getClass().getClassLoader()
+									: applicationContext.getClassLoader());
 			this.syncCommitTimeout = determineSyncCommitTimeout();
 			if (this.containerProperties.getSyncCommitTimeout() == null) {
 				// update the property so we can use it directly from code elsewhere
@@ -1247,27 +1254,6 @@ else if (timeout instanceof String str) {
 			}
 		}
 
-		@Nullable
-		private Object findDeserializerClass(Map<String, Object> props, Properties consumerOverrides, boolean isValue) {
-			Object configuredDeserializer = isValue
-					? KafkaMessageListenerContainer.this.consumerFactory.getValueDeserializer()
-					: KafkaMessageListenerContainer.this.consumerFactory.getKeyDeserializer();
-			if (configuredDeserializer == null) {
-				Object deser = consumerOverrides.get(isValue
-						? ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
-						: ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
-				if (deser == null) {
-					deser = props.get(isValue
-							? ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG
-							: ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
-				}
-				return deser;
-			}
-			else {
-				return configuredDeserializer.getClass();
-			}
-		}
-
 		private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> subscribingConsumer) {
 			if (KafkaMessageListenerContainer.this.topicPartitions == null) {
 				ConsumerRebalanceListener rebalanceListener = new ListenerConsumerRebalanceListener();
@@ -1293,29 +1279,6 @@ private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> subscr
 			}
 		}
 
-		private boolean checkDeserializer(@Nullable Object deser) {
-			Class<?> deserializer = null;
-			if (deser instanceof Class<?> deserClass) {
-				deserializer = deserClass;
-			}
-			else if (deser instanceof String str) {
-				try {
-					ApplicationContext applicationContext = getApplicationContext();
-					ClassLoader classLoader = applicationContext == null
-							? getClass().getClassLoader()
-							: applicationContext.getClassLoader();
-					deserializer = ClassUtils.forName(str, classLoader);
-				}
-				catch (ClassNotFoundException | LinkageError e) {
-					throw new IllegalStateException(e);
-				}
-			}
-			else if (deser != null) {
-				throw new IllegalStateException("Deserializer must be a class or class name, not a " + deser.getClass());
-			}
-			return deserializer != null && ErrorHandlingDeserializer.class.isAssignableFrom(deserializer);
-		}
-
 		protected void checkConsumer() {
 			long timeSinceLastPoll = System.currentTimeMillis() - this.lastPoll;
 			if (((float) timeSinceLastPoll) / (float) this.containerProperties.getPollTimeout()