GH-2638: Add Dynamic Micrometer Tags

* GH-2638: Add Dynamic Micrometer Tags

Resolves #2638

**cherry-pick to 2.9.x**

TODO: Support for Observations when `observationEnabled` (3.0.x only).

* Fix docs.
# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
#	spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
#	spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
#	spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/MicrometerHolder.java

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2590,6 +2590,14 @@ Also see `idleBeforeDataMultiplier`.
 |`true`
 |Whether or not to maintain Micrometer timers for the consumer threads.
 
+|[[micrometerTags]]<<micrometerTags,`micrometerTags`>>
+|empty
+|A map of static tags to be added to micrometer metrics.
+
+|[[micrometerTagsProvider]]<<micrometerTagsProvider,`micrometerTagsProvider`>>
+|`null`
+|A function that provides dynamic tags, based on the consumer record.
+
 |[[missingTopicsFatal]]<<missingTopicsFatal,`missingTopicsFatal`>>
 |`false`
 |When true prevents the container from starting if the confifgured topic(s) are not present on the broker.
@@ -3401,6 +3409,8 @@ The timers are named `spring.kafka.listener` and have the following tags:
 
 You can add additional tags using the `ContainerProperties` `micrometerTags` property.
 
+Starting with versions 2.9.8, 3.0.6, you can provide a function in `ContainerProperties` `micrometerTagsProvider`; the function receives the `ConsumerRecord<?, ?>` and returns tags which can be based on that record, and merged with any static tags in `micrometerTags`.
+
 NOTE: With the concurrent container, timers are created for each thread and the `name` tag is suffixed with `-n` where n is `0` to `concurrency-1`.
 
 ===== Monitoring KafkaTemplate Performance
@@ -3418,6 +3428,8 @@ The timers are named `spring.kafka.template` and have the following tags:
 
 You can add additional tags using the template's `micrometerTags` property.
 
+Starting with versions 2.9.8, 3.0.6, you can provide a `KafkaTemplate.setMicrometerTagsProvider(Function<ProducerRecord<?, ?>, Map<String, String>>)` property; the function receives the `ProducerRecord<?, ?>` and returns tags which can be based on that record, and merged with any static tags in `micrometerTags`.
+
 [[micrometer-native]]
 ===== Micrometer Native Metrics
 

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -27,6 +27,7 @@
 import java.util.Properties;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.function.Function;
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.Consumer;
@@ -130,6 +131,8 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private volatile MicrometerHolder micrometerHolder;
 
+	@Nullable
+	private Function<ProducerRecord<?, ?>, Map<String, String>> micrometerTagsProvider;
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
 	 * @param producerFactory the producer factory.
@@ -341,6 +344,33 @@ public void setMicrometerTags(Map<String, String> tags) {
 		}
 	}
 
+	/**
+	 * Set a function to provide dynamic tags based on the producer record. These tags
+	 * will be added to any static tags provided in {@link #setMicrometerTags(Map)
+	 * micrometerTags}. Only applies to record listeners, ignored for batch listeners.
+	 * Does not apply if observation is enabled.
+	 * @param micrometerTagsProvider the micrometerTagsProvider.
+	 * @since 2.9.8
+	 * @see #setMicrometerEnabled(boolean)
+	 * @see #setMicrometerTags(Map)
+	 * @see #setObservationEnabled(boolean)
+	 */
+	public void setMicrometerTagsProvider(
+			@Nullable Function<ProducerRecord<?, ?>, Map<String, String>> micrometerTagsProvider) {
+
+		this.micrometerTagsProvider = micrometerTagsProvider;
+	}
+
+	/**
+	 * Return the Micrometer tags provider.
+	 * @return the micrometerTagsProvider.
+	 * @since 2.9.8
+	 */
+	@Nullable
+	public Function<ProducerRecord<?, ?>, Map<String, String>> getMicrometerTagsProvider() {
+		return this.micrometerTagsProvider;
+	}
+
 	/**
 	 * Return the producer factory used by this template.
 	 * @return the factory.
@@ -680,9 +710,7 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 		return (metadata, exception) -> {
 			try {
 				if (exception == null) {
-					if (sample != null) {
-						this.micrometerHolder.success(sample);
-					}
+					successTimer(sample, producerRecord);
 					future.set(new SendResult<>(producerRecord, metadata));
 					if (KafkaTemplate.this.producerListener != null) {
 						KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
@@ -691,9 +719,7 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 							+ ", metadata: " + metadata);
 				}
 				else {
-					if (sample != null) {
-						this.micrometerHolder.failure(sample, exception.getClass().getSimpleName());
-					}
+					failureTimer(sample, exception, producerRecord);
 					future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
 					if (KafkaTemplate.this.producerListener != null) {
 						KafkaTemplate.this.producerListener.onError(producerRecord, metadata, exception);
@@ -710,6 +736,28 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 		};
 	}
 
+	private void successTimer(@Nullable Object sample, ProducerRecord<?, ?> record) {
+		if (sample != null) {
+			if (this.micrometerTagsProvider == null) {
+				this.micrometerHolder.success(sample);
+			}
+			else {
+				this.micrometerHolder.success(sample, record);
+			}
+		}
+	}
+
+	private void failureTimer(@Nullable Object sample, Exception exception, ProducerRecord<?, ?> record) {
+		if (sample != null) {
+			if (this.micrometerTagsProvider == null) {
+				this.micrometerHolder.failure(sample, exception.getClass().getSimpleName());
+			}
+			else {
+				this.micrometerHolder.failure(sample, exception.getClass().getSimpleName(), record);
+			}
+		}
+	}
+
 
 	/**
 	 * Return true if the template is currently running in a transaction on the calling
@@ -767,9 +815,18 @@ private MicrometerHolder obtainMicrometerHolder() {
 		MicrometerHolder holder = null;
 		try {
 			if (KafkaUtils.MICROMETER_PRESENT) {
+				Function<Object, Map<String, String>> mergedProvider = cr -> this.micrometerTags;
+				if (this.micrometerTagsProvider != null) {
+					mergedProvider = cr -> {
+						Map<String, String> tags = new HashMap<>(this.micrometerTags);
+						if (cr != null) {
+							tags.putAll(this.micrometerTagsProvider.apply((ProducerRecord<?, ?>) cr));
+						}
+						return tags;
+					};
+				}
 				holder = new MicrometerHolder(this.applicationContext, this.beanName,
-						"spring.kafka.template", "KafkaTemplate Timer",
-						this.micrometerTags);
+						"spring.kafka.template", "KafkaTemplate Timer", mergedProvider);
 			}
 		}
 		catch (@SuppressWarnings("unused") IllegalStateException ex) {

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

@@ -23,9 +23,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 
 import org.aopalliance.aop.Advice;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.framework.ProxyFactory;
@@ -229,6 +231,9 @@ public EOSMode getMode() {
 
 	private final List<Advice> adviceChain = new ArrayList<>();
 
+	@Nullable
+	private Function<ConsumerRecord<?, ?>, Map<String, String>> micrometerTagsProvider;
+
 	/**
 	 * The ack mode to use when auto ack (in the configuration properties) is false.
 	 * <ul>
@@ -732,10 +737,42 @@ public void setMicrometerTags(Map<String, String> tags) {
 		}
 	}
 
+	/**
+	 * Return static Micrometer tags.
+	 * @return the tags.
+	 * @since 2.3
+	 */
 	public Map<String, String> getMicrometerTags() {
 		return Collections.unmodifiableMap(this.micrometerTags);
 	}
 
+	/**
+	 * Set a function to provide dynamic tags based on the consumer record. These tags
+	 * will be added to any static tags provided in {@link #setMicrometerTags(Map)
+	 * micrometerTags}. Only applies to record listeners, ignored for batch listeners.
+	 * Does not apply if observation is enabled.
+	 * @param micrometerTagsProvider the micrometerTagsProvider.
+	 * @since 2.9.8
+	 * @see #setMicrometerEnabled(boolean)
+	 * @see #setMicrometerTags(Map)
+	 * @see #setObservationEnabled(boolean)
+	 */
+	public void setMicrometerTagsProvider(
+			@Nullable Function<ConsumerRecord<?, ?>, Map<String, String>> micrometerTagsProvider) {
+
+		this.micrometerTagsProvider = micrometerTagsProvider;
+	}
+
+	/**
+	 * Return the Micrometer tags provider.
+	 * @return the micrometerTagsProvider.
+	 * @since 2.9.8
+	 */
+	@Nullable
+	public Function<ConsumerRecord<?, ?>, Map<String, String>> getMicrometerTagsProvider() {
+		return this.micrometerTagsProvider;
+	}
+
 	public Duration getConsumerStartTimeout() {
 		return this.consumerStartTimeout;
 	}

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -42,6 +42,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
@@ -752,6 +753,9 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final boolean pauseImmediate = this.containerProperties.isPauseImmediate();
 
+		@Nullable
+		private final Function<ConsumerRecord<?, ?>, Map<String, String>> micrometerTagsProvider =
+				this.containerProperties.getMicrometerTagsProvider();
 		private Map<TopicPartition, OffsetMetadata> definedPartitions;
 
 		private int count;
@@ -1230,9 +1234,19 @@ private MicrometerHolder obtainMicrometerHolder() {
 			MicrometerHolder holder = null;
 			try {
 				if (KafkaUtils.MICROMETER_PRESENT && this.containerProperties.isMicrometerEnabled()) {
+					Function<Object, Map<String, String>> mergedProvider =
+							cr -> this.containerProperties.getMicrometerTags();
+					if (this.micrometerTagsProvider != null) {
+						mergedProvider = cr -> {
+							Map<String, String> tags = new HashMap<>(this.containerProperties.getMicrometerTags());
+							if (cr != null) {
+								tags.putAll(this.micrometerTagsProvider.apply((ConsumerRecord<?, ?>) cr));
+							}
+							return tags;
+						};
+					}
 					holder = new MicrometerHolder(getApplicationContext(), getBeanName(),
-							"spring.kafka.listener", "Kafka Listener Timer",
-							this.containerProperties.getMicrometerTags());
+							"spring.kafka.listener", "Kafka Listener Timer", mergedProvider);
 				}
 			}
 			catch (@SuppressWarnings(UNUSED) IllegalStateException ex) {
@@ -2211,7 +2225,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 			try {
 				invokeBatchOnMessage(records, recordList);
 				batchInterceptAfter(records, null);
-				successTimer(sample);
+				successTimer(sample, null);
 				if (this.batchFailed) {
 					this.batchFailed = false;
 					if (this.commonErrorHandler != null) {
@@ -2224,7 +2238,7 @@ private RuntimeException doInvokeBatchListener(final ConsumerRecords<K, V> recor
 				}
 			}
 			catch (RuntimeException e) {
-				failureTimer(sample);
+				failureTimer(sample, null);
 				batchInterceptAfter(records, e);
 				if (this.commonErrorHandler == null) {
 					throw e;
@@ -2302,15 +2316,25 @@ private Object startMicrometerSample() {
 			return null;
 		}
 
-		private void successTimer(@Nullable Object sample) {
+		private void successTimer(@Nullable Object sample, @Nullable ConsumerRecord<?, ?> record) {
 			if (sample != null) {
-				this.micrometerHolder.success(sample);
+				if (this.micrometerTagsProvider == null || record == null) {
+					this.micrometerHolder.success(sample);
+				}
+				else {
+					this.micrometerHolder.success(sample, record);
+				}
 			}
 		}
 
-		private void failureTimer(@Nullable Object sample) {
+		private void failureTimer(@Nullable Object sample, @Nullable ConsumerRecord<?, ?> record) {
 			if (sample != null) {
-				this.micrometerHolder.failure(sample, "ListenerExecutionFailedException");
+				if (this.micrometerTagsProvider == null || record == null) {
+					this.micrometerHolder.failure(sample, "ListenerExecutionFailedException");
+				}
+				else {
+					this.micrometerHolder.failure(sample, "ListenerExecutionFailedException", record);
+				}
 			}
 		}
 
@@ -2698,11 +2722,11 @@ private RuntimeException doInvokeRecordListener(final ConsumerRecord<K, V> recor
 
 			try {
 				invokeOnMessage(record);
-				successTimer(sample);
+				successTimer(sample, record);
 				recordInterceptAfter(record, null);
 			}
 			catch (RuntimeException e) {
-				failureTimer(sample);
+				failureTimer(sample, record);
 				recordInterceptAfter(record, e);
 				if (this.commonErrorHandler == null) {
 					throw e;