From da9dc280508f97b4678a7314133fbd9104d3c301 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 1 Oct 2019 09:55:13 -0400 Subject: [PATCH] GH-1225: @KL - skip conversion when not needed 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...(). --- .../BatchMessagingMessageListenerAdapter.java | 3 - .../MessagingMessageListenerAdapter.java | 69 +++++++++++++------ ...RecordMessagingMessageListenerAdapter.java | 16 ++++- .../EnableKafkaIntegrationTests.java | 47 +++++++++++-- 4 files changed, 102 insertions(+), 33 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java index 7aa9577f..311833e0 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/BatchMessagingMessageListenerAdapter.java @@ -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 extends MessagingMessageListenerAdapter implements BatchAcknowledgingConsumerAwareMessageListener { - private static final Message NULL_MESSAGE = new GenericMessage<>(KafkaNull.INSTANCE); - private BatchMessageConverter batchMessageConverter = new BatchMessagingMessageConverter(); private KafkaListenerErrorHandler errorHandler; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java index c9ca372c..a86f9759 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java @@ -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 implements ConsumerS private static final ParserContext PARSER_CONTEXT = new TemplateParserContext("!{", "}"); + /** + * Message used when no conversion is needed. + */ + protected static final Message 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 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 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 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 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 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 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 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); } /** diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java index 714f245a..8eb521ce 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordMessagingMessageListenerAdapter.java @@ -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 extends MessagingMessag */ @Override public void onMessage(ConsumerRecord 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 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); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index a60bae22..2635efe4 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -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> + factoryWithBadConverter() { + + ConcurrentKafkaListenerContainerFactory 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> 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(); }