diff --git a/spring-kafka-docs/src/main/asciidoc/kafka.adoc b/spring-kafka-docs/src/main/asciidoc/kafka.adoc index 01396260..84de44cb 100644 --- a/spring-kafka-docs/src/main/asciidoc/kafka.adoc +++ b/spring-kafka-docs/src/main/asciidoc/kafka.adoc @@ -3447,7 +3447,11 @@ The default implementations add the `bean.name` tag for template observations an You can either subclass `DefaultKafkaTemplateObservationConvention` or `DefaultKafkaListenerObservationConvention` or provide completely new implementations. -See <> for details of the observations that are recorded. +See <> for details of the default observations that are recorded. + +Starting with version 3.0.6, you can add dynamic tags to the timers and traces, based on information in the consumer or producer records. +To do so, add a custom `KafkaListenerObservationConvention` and/or `KafkaTemplateObservationConvention` to the listener container properties or `KafkaTemplate` respectively. +The `record` property in both observation contexts contains the `ConsumerRecord` or `ProducerRecord` respectively. [[transactions]] ==== Transactions diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java index dbd62574..016dd358 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java @@ -52,6 +52,10 @@ public class KafkaRecordReceiverContext extends ReceiverContext getRecord() { + return this.record; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java index c56953aa..c8cfd3ae 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java @@ -34,17 +34,21 @@ public class KafkaRecordSenderContext extends SenderContext private final String beanName; - private final String destination; + private final ProducerRecord record; public KafkaRecordSenderContext(ProducerRecord record, String beanName, Supplier clusterId) { super((carrier, key, value) -> record.headers().add(key, value.getBytes(StandardCharsets.UTF_8))); setCarrier(record); this.beanName = beanName; - this.destination = record.topic(); + this.record = record; String cluster = clusterId.get(); setRemoteServiceName("Apache Kafka" + (cluster != null ? ": " + cluster : "")); } + /** + * Return the template's bean name. + * @return the name. + */ public String getBeanName() { return this.beanName; } @@ -54,7 +58,16 @@ public class KafkaRecordSenderContext extends SenderContext * @return the topic. */ public String getDestination() { - return this.destination; + return this.record.topic(); + } + + /** + * Return the producer record. + * @return the record the record. + * @since 3.0.6 + */ + public ProducerRecord getRecord() { + return this.record; } }