GH-1479: ConsumerRecordMetadata

Resolves #1479

Because we pass the payload in as a discrete argument, standard headers
such as `RECEIVED_TOPIC` will get the payload value if it's a String.
See spring-projects/spring-framework#25033

Provide a mechanism to get the record metadata as a parameter instead.
Make this available in all listeners for consistency.

* Polishing

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/AdapterUtils.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener.adapter;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+
+import org.springframework.lang.Nullable;
+
+/**
+ * Utilities for listener adapters.
+ *
+ * @author Gary Russell
+ * @since 2.5
+ *
+ */
+public final class AdapterUtils {
+
+	private AdapterUtils() {
+	}
+
+	/**
+	 * Build a {@link ConsumerRecordMetadata} from the first {@link ConsumerRecord} in data, if any.
+	 * @param data the data array.
+	 * @return the metadata or null if data does not contain a {@link ConsumerRecord}.
+	 */
+	@Nullable
+	public static Object buildConsumerRecordMetadataFromArray(Object... data) {
+		for (Object object : data) {
+			ConsumerRecordMetadata metadata = buildConsumerRecordMetadata(object);
+			if (metadata != null) {
+				return metadata;
+			}
+		}
+		return null;
+	}
+
+	/**
+	 * Build a {@link ConsumerRecordMetadata} from data which must be a
+	 * {@link ConsumerRecord}.
+	 * @param data the record.
+	 * @return the metadata or null if data is not a {@link ConsumerRecord}.
+	 */
+	@Nullable
+	public static ConsumerRecordMetadata buildConsumerRecordMetadata(Object data) {
+		if (!(data instanceof ConsumerRecord)) {
+			return null;
+		}
+		ConsumerRecord<?, ?> record = (ConsumerRecord<?, ?>) data;
+		return new ConsumerRecordMetadata(new RecordMetadata(new TopicPartition(record.topic(), record.partition()),
+				0, record.offset(), record.timestamp(), null, record.serializedKeySize(),
+				record.serializedValueSize()), record.timestampType());
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/ConsumerRecordMetadata.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener.adapter;
+
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.record.TimestampType;
+
+/**
+ * Used to provide a listener method argument when the user supplies such a parameter.
+ * Delegates to {@link RecordMetadata} (which is final, hence no subclass) for all except
+ * timestamp type.
+ *
+ * @author Gary Russell
+ * @since 2.5
+ *
+ */
+public class ConsumerRecordMetadata {
+
+	private final RecordMetadata delegate;
+
+	private final TimestampType timestampType;
+
+	public ConsumerRecordMetadata(RecordMetadata delegate, TimestampType timestampType) {
+		this.delegate = delegate;
+		this.timestampType = timestampType;
+	}
+
+	public boolean hasOffset() {
+		return this.delegate.hasOffset();
+	}
+
+	public long offset() {
+		return this.delegate.offset();
+	}
+
+	public boolean hasTimestamp() {
+		return this.delegate.hasTimestamp();
+	}
+
+	public long timestamp() {
+		return this.delegate.timestamp();
+	}
+
+	public int serializedKeySize() {
+		return this.delegate.serializedKeySize();
+	}
+
+	public int serializedValueSize() {
+		return this.delegate.serializedValueSize();
+	}
+
+	public String topic() {
+		return this.delegate.topic();
+	}
+
+	public int partition() {
+		return this.delegate.partition();
+	}
+
+	public TimestampType timestampType() {
+		return this.timestampType;
+	}
+
+	@Override
+	public int hashCode() {
+		return this.delegate.hashCode() + this.timestampType.name().hashCode();
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (!(obj instanceof ConsumerRecordMetadata)) {
+			return false;
+		}
+		return this.delegate.equals(obj)
+				&& this.timestampType.equals(((ConsumerRecordMetadata) obj).timestampType());
+	}
+
+	@Override
+	public String toString() {
+		return this.delegate.toString();
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
+import java.lang.reflect.Parameter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -40,6 +41,7 @@
 import org.springframework.kafka.support.KafkaUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+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.invocation.InvocableHandlerMethod;
@@ -123,8 +125,11 @@ public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 			Object bean, BeanExpressionResolver beanExpressionResolver, BeanExpressionContext beanExpressionContext,
 			@Nullable BeanFactory beanFactory) {
 
-		this.handlers = new ArrayList<>(handlers);
-		this.defaultHandler = defaultHandler;
+		this.handlers = new ArrayList<>();
+		for (InvocableHandlerMethod handler : handlers) {
+			this.handlers.add(wrapIfNecessary(handler));
+		}
+		this.defaultHandler = wrapIfNecessary(defaultHandler);
 		this.bean = bean;
 		this.resolver = beanExpressionResolver;
 		this.beanExpressionContext = beanExpressionContext;
@@ -133,6 +138,19 @@ public DelegatingInvocableHandler(List<InvocableHandlerMethod> handlers,
 				: null;
 	}
 
+	private InvocableHandlerMethod wrapIfNecessary(InvocableHandlerMethod handler) {
+		if (handler == null) {
+			return null;
+		}
+		Parameter[] parameters = handler.getMethod().getParameters();
+		for (Parameter parameter : parameters) {
+			if (parameter.getType().equals(ConsumerRecordMetadata.class)) {
+				return new DelegatingInvocableHandler.MetadataAwareInvocableHandlerMethod(handler);
+			}
+		}
+		return handler;
+	}
+
 	/**
 	 * Return the bean for this handler.
 	 * @return the bean.
@@ -152,7 +170,16 @@ public Object getBean() {
 	public Object invoke(Message<?> message, Object... providedArgs) throws Exception { //NOSONAR
 		Class<? extends Object> payloadClass = message.getPayload().getClass();
 		InvocableHandlerMethod handler = getHandlerForPayload(payloadClass);
-		Object result = handler.invoke(message, providedArgs);
+		Object result;
+		if (handler instanceof MetadataAwareInvocableHandlerMethod) {
+			Object[] args = new Object[providedArgs.length + 1];
+			args[0] = AdapterUtils.buildConsumerRecordMetadataFromArray(providedArgs);
+			System.arraycopy(providedArgs, 0, args, 1, providedArgs.length);
+			result = handler.invoke(message, args);
+		}
+		else {
+			result = handler.invoke(message, providedArgs);
+		}
 		Expression replyTo = this.handlerSendTo.get(handler);
 		return new InvocationResult(result, replyTo, this.handlerReturnsMessage.get(handler));
 	}
@@ -284,4 +311,17 @@ 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);
+		}
+
+	}
+
 }

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

@@ -120,6 +120,8 @@ public abstract class MessagingMessageListenerAdapter<K, V> implements ConsumerS
 
 	private boolean hasAckParameter;
 
+	private boolean hasMetadataParameter;
+
 	private boolean messageReturnType;
 
 	private ReplyHeadersConfigurer replyHeadersConfigurer;
@@ -324,7 +326,13 @@ protected final Object invokeHandler(Object data, Acknowledgment acknowledgment,
 				return this.handlerMethod.invoke(message, acknowledgment, consumer);
 			}
 			else {
-				return this.handlerMethod.invoke(message, data, acknowledgment, consumer);
+				if (this.hasMetadataParameter) {
+					return this.handlerMethod.invoke(message, data, acknowledgment, consumer,
+							AdapterUtils.buildConsumerRecordMetadata(data));
+				}
+				else {
+					return this.handlerMethod.invoke(message, data, acknowledgment, consumer);
+				}
 			}
 		}
 		catch (org.springframework.messaging.converter.MessageConversionException ex) {
@@ -562,6 +570,9 @@ protected Type determineInferredType(Method method) { // NOSONAR complexity
 			isNotConvertible |= isAck;
 			boolean isConsumer = parameterIsType(parameterType, Consumer.class);
 			isNotConvertible |= isConsumer;
+			boolean isMeta = parameterIsType(parameterType, ConsumerRecordMetadata.class);
+			this.hasMetadataParameter |= isMeta;
+			isNotConvertible |= isMeta;
 			if (isNotConvertible) {
 				notConvertibleParameters++;
 			}