Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Refactor DeserializationException Detection Code #2756

Merged
merged 2 commits into from
Jul 26, 2023
Merged

Refactor DeserializationException Detection Code #2756

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ddd1a96
Refactor DeserializationException Detection Code
garyrussell Jul 26, 2023
89caea5
Pass class loader into utility method instead of an ApplicationContext.
garyrussell Jul 26, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion ...ka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
Expand Down Expand Up @@ -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;
Expand Down
70 changes: 70 additions & 0 deletions spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,27 @@

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;
import org.apache.kafka.common.errors.WakeupException;

import org.springframework.classify.BinaryExceptionClassifier;
import org.springframework.context.ApplicationContext;
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;

Expand Down Expand Up @@ -256,4 +264,66 @@ 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 applicationContext the application context.
* @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, @Nullable ApplicationContext applicationContext) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't have an application context in my hands, but I do configure a ClassLoader.
How about to change it to that one instead, please?


Object deser = findDeserializerClass(consumerFactory, consumerOverrides, isValue);
Class<?> deserializer = null;
if (deser instanceof Class<?> deserClass) {
deserializer = deserClass;
}
else if (deser instanceof String str) {
try {
ClassLoader classLoader = applicationContext == null
? consumerFactory.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);
}

@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();
}
}

}
52 changes: 4 additions & 48 deletions ...kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -920,9 +918,11 @@ else if (listener instanceof MessageListener) {
}
Map<String, Object> props = KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties();
this.checkNullKeyForExceptions = this.containerProperties.isCheckDeserExWhenKeyNull()
|| checkDeserializer(findDeserializerClass(props, consumerProperties, false));
|| ErrorHandlingUtils.checkDeserializer(KafkaMessageListenerContainer.this.consumerFactory, consumerProperties, false,
getApplicationContext());
this.checkNullValueForExceptions = this.containerProperties.isCheckDeserExWhenValueNull()
|| checkDeserializer(findDeserializerClass(props, consumerProperties, true));
|| ErrorHandlingUtils.checkDeserializer(KafkaMessageListenerContainer.this.consumerFactory, consumerProperties, true,
getApplicationContext());
this.syncCommitTimeout = determineSyncCommitTimeout();
if (this.containerProperties.getSyncCommitTimeout() == null) {
// update the property so we can use it directly from code elsewhere
Expand Down Expand Up @@ -1247,27 +1247,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();
Expand All @@ -1293,29 +1272,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()
Expand Down
Loading