Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2638: Support Dynamic Tags via Observation

Browse Source 
See https://github.com/spring-projects/spring-kafka/pull/2648
This commit is contained in:
Gary Russell
2023-03-30 16:35:05 -04:00
committed by GitHub
parent 0f8dc8055b
commit f351a7a4aa
3 changed files with 34 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-kafka-docs/src/main/asciidoc/kafka.adoc
View File

@@ -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 <<observation-gen>> for details of the observations that are recorded.
See <<observation-gen>> 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

13
spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordReceiverContext.java
View File

@@ -52,6 +52,10 @@ public class KafkaRecordReceiverContext extends ReceiverContext<ConsumerRecord<?
setRemoteServiceName("Apache Kafka" + (cluster != null ? ": " + cluster : ""));
}
/**
* Return the listener id.
* @return the listener id.
*/
public String getListenerId() {
return this.listenerId;
}
@@ -64,4 +68,13 @@ public class KafkaRecordReceiverContext extends ReceiverContext<ConsumerRecord<?
return this.record.topic();
}
/**
* Return the consumer record.
* @return the record the record.
* @since 3.0.6
*/
public ConsumerRecord<?, ?> getRecord() {
return this.record;
}
}

19
spring-kafka/src/main/java/org/springframework/kafka/support/micrometer/KafkaRecordSenderContext.java
View File

@@ -34,17 +34,21 @@ public class KafkaRecordSenderContext extends SenderContext<ProducerRecord<?, ?>
private final String beanName;
private final String destination;
private final ProducerRecord<?, ?> record;
public KafkaRecordSenderContext(ProducerRecord<?, ?> record, String beanName, Supplier<String> 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<ProducerRecord<?, ?>
* @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;
}
}