GH-2295: No Resolvers with ConsumerRecordMetadata

Resolves #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**

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/DelegatingInvocableHandler.java

@@ -40,7 +40,6 @@
 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<InvocableHandlerMethod, Boolean> handlerReturnsMessage = new ConcurrentHashMap<>();
 
+	private final Map<InvocableHandlerMethod, Boolean> handlerMetadataAware = new ConcurrentHashMap<>();
+
 	private final Object bean;
 
 	private final BeanExpressionResolver resolver;
@@ -102,11 +103,12 @@ public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 			@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;
@@ -116,18 +118,17 @@ public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 		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;
 	}
 
 	/**
@@ -156,7 +157,7 @@ public Object invoke(Message<?> message, Object... providedArgs) throws Exceptio
 			}
 		}
 		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);
@@ -315,19 +316,6 @@ public boolean hasDefaultHandler() {
 		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 for: spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -108,6 +108,7 @@
 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;
@@ -447,6 +448,7 @@ public void testMulti() throws Exception {
 
 		template.send("annotated8", 0, 1, "junk");
 		assertThat(this.multiListener.errorLatch.await(60, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.multiListener.meta).isNotNull();
 	}
 
 	@Test
@@ -2302,18 +2304,21 @@ public CountDownLatch getLatch2() {
 	@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);
+
+		final CountDownLatch errorLatch = new CountDownLatch(1);
 
-		private 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();
 			}
 		}