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