Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1225: @KL - skip conversion when not needed

Browse Source 
Resolves https://github.com/spring-projects/spring-kafka/issues/1225

When no argument is sourced from a `Message<?>` (no `@Headers`, `@Payload`,
etc - only `ConsumerRecord`, `Acknowledgment` or `Consumer`) then don't
convert.

* Fix logging - don't use supplier with is...().
This commit is contained in:
Gary Russell
2019-10-01 09:55:13 -04:00
committed by Artem Bilan
parent f86110e831
commit da9dc28050
4 changed files with 102 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java
View File

@@ -28,7 +28,6 @@ import org.springframework.kafka.listener.BatchAcknowledgingConsumerAwareMessage
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.messaging.Message;
@@ -59,8 +58,6 @@ import org.springframework.messaging.support.MessageBuilder;
public class BatchMessagingMessageListenerAdapter<K, V> extends MessagingMessageListenerAdapter<K, V>
implements BatchAcknowledgingConsumerAwareMessageListener<K, V> {
private static final Message<KafkaNull> NULL_MESSAGE = new GenericMessage<>(KafkaNull.INSTANCE);
private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter();
private KafkaListenerErrorHandler errorHandler;

69
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -49,6 +49,7 @@ import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.KafkaUtils;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -59,6 +60,7 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -83,6 +85,11 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private static final ParserContext PARSER_CONTEXT = new TemplateParserContext("!{", "}");
/**
* Message used when no conversion is needed.
*/
protected static final Message<KafkaNull> NULL_MESSAGE = new GenericMessage<>(KafkaNull.INSTANCE); // NOSONAR
private final Object bean;
protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); //NOSONAR
@@ -99,6 +106,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
private boolean isMessageList;
private boolean conversionNeeded = true;
private RecordMessageConverter messageConverter = new MessagingMessageConverter();
private Type fallbackType = Object.class;
@@ -174,6 +183,10 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
return this.isConsumerRecords;
}
public boolean isConversionNeeded() {
return this.conversionNeeded;
}
/**
* Set the topic to which to send any result from the method invocation.
* May be a SpEL expression {@code !{...}} evaluated at runtime.
@@ -481,14 +494,26 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
Type genericParameterType = null;
int allowedBatchParameters = 1;
int notConvertibleParameters = 0;
for (int i = 0; i < method.getParameterCount(); i++) {
MethodParameter methodParameter = new MethodParameter(method, i);
/*
* We're looking for a single non-annotated parameter, or one annotated with @Payload.
* We ignore parameters with type Message because they are not involved with conversion.
* We ignore parameters with type Message, Consumer, Ack, ConsumerRecord because they
* are not involved with conversion.
*/
if (eligibleParameter(methodParameter)
Type parameterType = methodParameter.getGenericParameterType();
boolean isNotConvertible = parameterIsType(parameterType, ConsumerRecord.class);
boolean isAck = parameterIsType(parameterType, Acknowledgment.class);
this.hasAckParameter |= isAck;
isNotConvertible |= isAck;
boolean isConsumer = parameterIsType(parameterType, Consumer.class);
isNotConvertible |= isConsumer;
if (isNotConvertible) {
notConvertibleParameters++;
}
if (!isNotConvertible && !isMessageWithNoTypeInfo(parameterType)
&& (methodParameter.getParameterAnnotations().length == 0
|| methodParameter.hasParameterAnnotation(Payload.class))) {
if (genericParameterType == null) {
@@ -500,8 +525,7 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
break;
}
}
else if (methodParameter.getGenericParameterType().equals(Acknowledgment.class)) {
this.hasAckParameter = true;
else if (isAck) {
allowedBatchParameters++;
}
else if (methodParameter.hasParameterAnnotation(Header.class)) {
@@ -511,11 +535,10 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
}
else {
if (methodParameter.getGenericParameterType().equals(Consumer.class)) {
if (isConsumer) {
allowedBatchParameters++;
}
else {
Type parameterType = methodParameter.getGenericParameterType();
if (parameterType instanceof ParameterizedType
&& ((ParameterizedType) parameterType).getRawType().equals(Consumer.class)) {
allowedBatchParameters++;
@@ -524,6 +547,9 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
}
}
if (notConvertibleParameters == method.getParameterCount()) {
this.conversionNeeded = false;
}
boolean validParametersForBatch = method.getGenericParameterTypes().length <= allowedBatchParameters;
if (!validParametersForBatch) {
@@ -587,27 +613,26 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
&& ((WildcardType) paramType).getUpperBounds().length > 0;
}
/*
* Don't consider parameter types that are available after conversion.
* Acknowledgment, ConsumerRecord, Consumer, ConsumerRecord<...>, Consumer<...>, and Message<?>.
*/
private boolean eligibleParameter(MethodParameter methodParameter) {
Type parameterType = methodParameter.getGenericParameterType();
if (parameterType.equals(Acknowledgment.class) || parameterType.equals(ConsumerRecord.class)
|| parameterType.equals(Consumer.class)) {
return false;
}
private boolean isMessageWithNoTypeInfo(Type parameterType) {
if (parameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) parameterType;
Type rawType = parameterizedType.getRawType();
if (rawType.equals(ConsumerRecord.class) || rawType.equals(Consumer.class)) {
return false;
}
else if (rawType.equals(Message.class)) {
return !(parameterizedType.getActualTypeArguments()[0] instanceof WildcardType);
if (rawType.equals(Message.class)) {
return parameterizedType.getActualTypeArguments()[0] instanceof WildcardType;
}
}
return !parameterType.equals(Message.class); // could be Message without a generic type
return parameterType.equals(Message.class); // could be Message without a generic type
}
private boolean parameterIsType(Type parameterType, Type type) {
if (parameterType instanceof ParameterizedType) {
ParameterizedType parameterizedType = (ParameterizedType) parameterType;
Type rawType = parameterizedType.getRawType();
if (rawType.equals(type)) {
return true;
}
}
return parameterType.equals(type);
}
/**

16
spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java
View File

@@ -26,6 +26,7 @@ import org.springframework.kafka.listener.KafkaListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -71,8 +72,16 @@ public class RecordMessagingMessageListenerAdapter<K, V> extends MessagingMessag
*/
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = toMessagingMessage(record, acknowledgment, consumer);
logger.debug(() -> "Processing [" + message + "]");
Message<?> message;
if (isConversionNeeded()) {
message = toMessagingMessage(record, acknowledgment, consumer);
}
else {
message = NULL_MESSAGE;
}
if (logger.isDebugEnabled()) {
logger.debug("Processing [" + message + "]");
}
try {
Object result = invokeHandler(record, acknowledgment, message, consumer);
if (result != null) {
@@ -82,6 +91,9 @@ public class RecordMessagingMessageListenerAdapter<K, V> extends MessagingMessag
catch (ListenerExecutionFailedException e) { // NOSONAR ex flow control
if (this.errorHandler != null) {
try {
if (message.equals(NULL_MESSAGE)) {
message = new GenericMessage<>(record);
}
Object result = this.errorHandler.handleError(message, e, consumer);
if (result != null) {
handleResult(result, record, message);

47
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -27,6 +27,7 @@ import static org.mockito.BDDMockito.willThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import java.lang.reflect.Type;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -53,6 +54,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.junit.jupiter.api.Test;
@@ -102,6 +104,7 @@ import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper;
import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
import org.springframework.kafka.support.converter.JsonMessageConverter;
import org.springframework.kafka.support.converter.ProjectingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
@@ -246,13 +249,13 @@ public class EnableKafkaIntegrationTests {
template.send("annotated3", 0, "foo");
template.flush();
assertThat(this.listener.latch3.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.record.value()).isEqualTo("foo");
assertThat(this.listener.capturedRecord.value()).isEqualTo("foo");
assertThat(this.config.listen3Exception).isNotNull();
template.send("annotated4", 0, "foo");
template.flush();
assertThat(this.listener.latch4.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.record.value()).isEqualTo("foo");
assertThat(this.listener.capturedRecord.value()).isEqualTo("foo");
assertThat(this.listener.ack).isNotNull();
assertThat(this.listener.eventLatch.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(this.listener.event.getListenerId().startsWith("qux-"));
@@ -847,6 +850,37 @@ public class EnableKafkaIntegrationTests {
return factory;
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
factoryWithBadConverter() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setRecordFilterStrategy(recordFilter());
factory.setReplyTemplate(partitionZeroReplyingTemplate());
factory.setErrorHandler((ConsumerAwareErrorHandler) (t, d, c) -> {
this.globalErrorThrowable = t;
c.seek(new org.apache.kafka.common.TopicPartition(d.topic(), d.partition()), d.offset());
});
factory.getContainerProperties().setMicrometerTags(Collections.singletonMap("extraTag", "foo"));
factory.setMessageConverter(new RecordMessageConverter() {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type payloadType) {
throw new UnsupportedOperationException();
}
@Override
public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
throw new UnsupportedOperationException(); }
});
return factory;
}
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
withNoReplyTemplateContainerFactory() {
@@ -1425,7 +1459,7 @@ public class EnableKafkaIntegrationTests {
volatile Integer partition;
volatile ConsumerRecord<?, ?> record;
volatile ConsumerRecord<?, ?> capturedRecord;
volatile Acknowledgment ack;
@@ -1502,12 +1536,13 @@ public class EnableKafkaIntegrationTests {
private final AtomicBoolean reposition3 = new AtomicBoolean();
@KafkaListener(id = "baz", topicPartitions = @TopicPartition(topic = "${topicThree:annotated3}",
partitions = "${zero:0}"), errorHandler = "listen3ErrorHandler")
partitions = "${zero:0}"), errorHandler = "listen3ErrorHandler",
containerFactory = "factoryWithBadConverter")
public void listen3(ConsumerRecord<?, ?> record) {
if (this.reposition3.compareAndSet(false, true)) {
throw new RuntimeException("reposition");
}
this.record = record;
this.capturedRecord = record;
this.latch3.countDown();
}
@@ -1537,7 +1572,7 @@ public class EnableKafkaIntegrationTests {
relativeToCurrent = "${zzz:true}"))
}, clientIdPrefix = "${foo.xxx:clientIdViaAnnotation}")
public void listen5(ConsumerRecord<?, ?> record) {
this.record = record;
this.capturedRecord = record;
this.latch5.countDown();
}