From 6416235c320cfcf6056b94b8a60157a38a172002 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 6 Jun 2022 15:34:08 -0400 Subject: [PATCH] GH-2295: No Resolvers with ConsumerRecordMetadata Resolves https://github.com/spring-projects/spring-kafka/issues/2295 A subclass of `InvocableHandlerMethod` was used to signal that a method needs a `ConsumerRecordMetadata` to be constructed. However, the constructor that takes an existing IHM, does not fully clone the argument passed into it. Then, the payload resolver was not invoked to resolve that parameter. Use a map of booleans instead, to signal the creation of the metadata. **cherry-pick to 2.9.x, 2.8.x, 2.7.x** --- .../adapter/DelegatingInvocableHandler.java | 34 ++++++------------- .../EnableKafkaIntegrationTests.java | 13 ++++--- 2 files changed, 20 insertions(+), 27 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java index abec5b9f..6066615f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java @@ -40,7 +40,6 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; -import org.springframework.messaging.handler.HandlerMethod; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver; @@ -75,6 +74,8 @@ public class DelegatingInvocableHandler { private final Map handlerReturnsMessage = new ConcurrentHashMap<>(); + private final Map handlerMetadataAware = new ConcurrentHashMap<>(); + private final Object bean; private final BeanExpressionResolver resolver; @@ -137,11 +138,12 @@ public class DelegatingInvocableHandler { @Nullable BeanExpressionContext beanExpressionContext, @Nullable BeanFactory beanFactory, @Nullable Validator validator) { - this.handlers = new ArrayList<>(); + this.handlers = new ArrayList<>(handlers); for (InvocableHandlerMethod handler : handlers) { - this.handlers.add(wrapIfNecessary(handler)); + checkSpecial(handler); } - this.defaultHandler = wrapIfNecessary(defaultHandler); + this.defaultHandler = defaultHandler; + checkSpecial(defaultHandler); this.bean = bean; this.resolver = beanExpressionResolver; this.beanExpressionContext = beanExpressionContext; @@ -151,18 +153,17 @@ public class DelegatingInvocableHandler { this.validator = validator == null ? null : new PayloadValidator(validator); } - @Nullable - private InvocableHandlerMethod wrapIfNecessary(@Nullable InvocableHandlerMethod handler) { + private void checkSpecial(@Nullable InvocableHandlerMethod handler) { if (handler == null) { - return null; + return; } Parameter[] parameters = handler.getMethod().getParameters(); for (Parameter parameter : parameters) { if (parameter.getType().equals(ConsumerRecordMetadata.class)) { - return new DelegatingInvocableHandler.MetadataAwareInvocableHandlerMethod(handler); + this.handlerMetadataAware.put(handler, true); + return; } } - return handler; } /** @@ -191,7 +192,7 @@ public class DelegatingInvocableHandler { } } Object result; - if (handler instanceof MetadataAwareInvocableHandlerMethod) { + if (Boolean.TRUE.equals(this.handlerMetadataAware.get(handler))) { Object[] args = new Object[providedArgs.length + 1]; args[0] = AdapterUtils.buildConsumerRecordMetadataFromArray(providedArgs); System.arraycopy(providedArgs, 0, args, 1, providedArgs.length); @@ -350,19 +351,6 @@ public class DelegatingInvocableHandler { return this.defaultHandler != null; } - /** - * A handler method that is aware of {@link ConsumerRecordMetadata}. - * - * @since 2.5 - */ - private static final class MetadataAwareInvocableHandlerMethod extends InvocableHandlerMethod { - - MetadataAwareInvocableHandlerMethod(HandlerMethod handlerMethod) { - super(handlerMethod); - } - - } - private static final class PayloadValidator extends PayloadMethodArgumentResolver { PayloadValidator(Validator validator) { 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 1545c674..54ed5598 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 @@ -107,6 +107,7 @@ import org.springframework.kafka.listener.KafkaListenerErrorHandler; import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.ListenerExecutionFailedException; import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata; import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter; import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; @@ -431,6 +432,7 @@ public class EnableKafkaIntegrationTests { template.send("annotated8", 0, 1, "junk"); assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue(); + assertThat(this.multiListener.meta).isNotNull(); } @Test @@ -2213,18 +2215,21 @@ public class EnableKafkaIntegrationTests { @KafkaListener(id = "multi", topics = "annotated8", errorHandler = "consumeMultiMethodException") static class MultiListenerBean { - private final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch1 = new CountDownLatch(1); - private final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); - private final CountDownLatch errorLatch = new CountDownLatch(1); + final CountDownLatch errorLatch = new CountDownLatch(1); + + volatile ConsumerRecordMetadata meta; @KafkaHandler - public void bar(@NonNull String bar) { + public void bar(@NonNull String bar, @Header(KafkaHeaders.RECORD_METADATA) ConsumerRecordMetadata meta) { if ("junk".equals(bar)) { throw new RuntimeException("intentional"); } else { + this.meta = meta; this.latch1.countDown(); } }