spring-projectsGH-2198: Switch from Timer to Observation

spring-projects#2198 Phase 1

Author: garyrussell

build.gradle

@@ -66,7 +66,8 @@ ext {
 	junitJupiterVersion = '5.8.2'
 	kafkaVersion = '3.1.0'
 	log4jVersion = '2.17.2'
-	micrometerVersion = '2.0.0-M3'
+	micrometerVersion = '2.0.0-SNAPSHOT'
+	micrometerTracingVersion = '1.0.0-SNAPSHOT'
 	mockitoVersion = '4.0.0'
 	reactorVersion = '2020.0.17'
 	scalaVersion = '2.13'
@@ -104,6 +105,8 @@ allprojects {
 			mavenBom "org.springframework:spring-framework-bom:$springVersion"
 			mavenBom "io.projectreactor:reactor-bom:$reactorVersion"
 			mavenBom "org.springframework.data:spring-data-bom:$springDataVersion"
+			mavenBom "io.micrometer:micrometer-bom:$micrometerVersion"
+			mavenBom "io.micrometer:micrometer-tracing-bom:$micrometerTracingVersion"
 		}
 	}
 
@@ -319,7 +322,6 @@ project ('spring-kafka') {
 		}
 		api "org.apache.kafka:kafka-clients:$kafkaVersion"
 		optionalApi "org.apache.kafka:kafka-streams:$kafkaVersion"
-
 		optionalApi 'com.fasterxml.jackson.core:jackson-core'
 		optionalApi 'com.fasterxml.jackson.core:jackson-databind'
 		optionalApi 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'
@@ -338,13 +340,16 @@ project ('spring-kafka') {
 
 		optionalApi 'io.projectreactor:reactor-core'
 		optionalApi 'io.projectreactor.kafka:reactor-kafka'
-		optionalApi "io.micrometer:micrometer-core:$micrometerVersion"
-        optionalApi "io.micrometer:micrometer-binders:$micrometerVersion"
+		optionalApi 'io.micrometer:micrometer-binders'
+		optionalApi 'io.micrometer:micrometer-tracing-api'
 
 		testImplementation project (':spring-kafka-test')
 		testImplementation 'io.projectreactor:reactor-test'
 		testImplementation "org.mockito:mockito-junit-jupiter:$mockitoVersion"
 		testImplementation "org.hibernate.validator:hibernate-validator:$hibernateValidationVersion"
+		testImplementation 'io.micrometer:micrometer-tracing-bridge-brave'
+		testImplementation 'io.micrometer:micrometer-tracing-test'
+		testImplementation 'io.micrometer:micrometer-tracing-integration-test'
 	}
 }
 

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

@@ -61,15 +61,21 @@
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
-import org.springframework.kafka.support.micrometer.MicrometerHolder;
+import org.springframework.kafka.support.micrometer.KafkaMetrics;
+import org.springframework.kafka.support.micrometer.KafkaTemplateObservation;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.SmartMessageConverter;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.concurrent.ListenableFuture;
 import org.springframework.util.concurrent.SettableListenableFuture;
 
+import io.micrometer.observation.Observation;
+import io.micrometer.observation.Observation.TagsProvider;
+import io.micrometer.observation.ObservationRegistry;
+
 
 /**
  * A template for executing high-level operations. When used with a
@@ -106,9 +112,9 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private final Map<String, String> micrometerTags = new HashMap<>();
 
-	private String beanName = "kafkaTemplate";
+	private final Observation.Context observationContext = new Observation.Context();
 
-	private ApplicationContext applicationContext;
+	private String beanName = "kafkaTemplate";
 
 	private RecordMessageConverter messageConverter = new MessagingMessageConverter();
 
@@ -126,11 +132,11 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V>, ApplicationCo
 
 	private ConsumerFactory<K, V> consumerFactory;
 
-	private volatile boolean micrometerEnabled = true;
+	private ProducerInterceptor<K, V> producerInterceptor;
 
-	private volatile MicrometerHolder micrometerHolder;
+	private TagsProvider<?> tagsProvider = KafkaMetrics.templateTagsProvider(this.micrometerTags);
 
-	private ProducerInterceptor<K, V> producerInterceptor;
+	private ObservationRegistry observationRegistry;
 
 	/**
 	 * Create an instance using the supplied producer factory and autoFlush false.
@@ -197,7 +203,6 @@ public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,
 
 		Assert.notNull(producerFactory, "'producerFactory' cannot be null");
 		this.autoFlush = autoFlush;
-		this.micrometerEnabled = KafkaUtils.MICROMETER_PRESENT;
 		this.customProducerFactory = configOverrides != null && configOverrides.size() > 0;
 		if (this.customProducerFactory) {
 			this.producerFactory = producerFactory.copyWithConfigurationOverride(configOverrides);
@@ -211,11 +216,12 @@ public KafkaTemplate(ProducerFactory<K, V> producerFactory, boolean autoFlush,
 	@Override
 	public void setBeanName(String name) {
 		this.beanName = name;
+		this.observationContext.put(KafkaTemplateObservation.TemplateLowCardinalityTags.BEAN_NAME.getKey(), name);
 	}
 
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) {
-		this.applicationContext = applicationContext;
+		this.observationRegistry = KafkaMetrics.getRegistry(applicationContext);
 		if (this.customProducerFactory) {
 			((DefaultKafkaProducerFactory<K, V>) this.producerFactory).setApplicationContext(applicationContext);
 		}
@@ -327,9 +333,10 @@ public boolean isAllowNonTransactional() {
 	 * Set to false to disable micrometer timers, if micrometer is on the class path.
 	 * @param micrometerEnabled false to disable.
 	 * @since 2.5
+	 * @deprecated this is no longer used.
 	 */
+	@Deprecated
 	public void setMicrometerEnabled(boolean micrometerEnabled) {
-		this.micrometerEnabled = micrometerEnabled;
 	}
 
 	/**
@@ -338,8 +345,9 @@ public void setMicrometerEnabled(boolean micrometerEnabled) {
 	 * @since 2.5
 	 */
 	public void setMicrometerTags(Map<String, String> tags) {
-		if (tags != null) {
+		if (!ObjectUtils.isEmpty(tags)) {
 			this.micrometerTags.putAll(tags);
+			this.tagsProvider = KafkaMetrics.templateTagsProvider(this.micrometerTags);
 		}
 	}
 
@@ -625,18 +633,19 @@ protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> p
 		final Producer<K, V> producer = getTheProducer(producerRecord.topic());
 		this.logger.trace(() -> "Sending: " + KafkaUtils.format(producerRecord));
 		final SettableListenableFuture<SendResult<K, V>> future = new SettableListenableFuture<>();
-		Object sample = null;
-		if (this.micrometerEnabled && this.micrometerHolder == null) {
-			this.micrometerHolder = obtainMicrometerHolder();
-		}
-		if (this.micrometerHolder != null) {
-			sample = this.micrometerHolder.start();
+		Observation observation = null;
+		if (this.observationRegistry != null) {
+			observation = Observation
+					.createNotStarted(KafkaTemplateObservation.TEMPLATE_OBSERVATION.getName(),
+							this.observationContext, this.observationRegistry)
+					.tagsProvider(this.tagsProvider)
+					.start();
 		}
 		if (this.producerInterceptor != null) {
 			this.producerInterceptor.onSend(producerRecord);
 		}
 		Future<RecordMetadata> sendFuture =
-				producer.send(producerRecord, buildCallback(producerRecord, producer, future, sample));
+				producer.send(producerRecord, buildCallback(producerRecord, producer, future, observation));
 		// May be an immediate failure
 		if (sendFuture.isDone()) {
 			try {
@@ -658,7 +667,7 @@ protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> p
 	}
 
 	private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final Producer<K, V> producer,
-			final SettableListenableFuture<SendResult<K, V>> future, @Nullable Object sample) {
+			final SettableListenableFuture<SendResult<K, V>> future, @Nullable Observation observation) {
 
 		return (metadata, exception) -> {
 			try {
@@ -671,9 +680,6 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 			}
 			try {
 				if (exception == null) {
-					if (sample != null) {
-						this.micrometerHolder.success(sample);
-					}
 					future.set(new SendResult<>(producerRecord, metadata));
 					if (KafkaTemplate.this.producerListener != null) {
 						KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
@@ -682,8 +688,8 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 							+ ", metadata: " + metadata);
 				}
 				else {
-					if (sample != null) {
-						this.micrometerHolder.failure(sample, exception.getClass().getSimpleName());
+					if (observation != null) {
+						observation.error(exception);
 					}
 					future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
 					if (KafkaTemplate.this.producerListener != null) {
@@ -694,6 +700,9 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 				}
 			}
 			finally {
+				if (observation != null) {
+					observation.stop();
+				}
 				if (!KafkaTemplate.this.transactional) {
 					closeProducer(producer, false);
 				}
@@ -753,27 +762,8 @@ else if (topic == null) {
 		}
 	}
 
-	@Nullable
-	private MicrometerHolder obtainMicrometerHolder() {
-		MicrometerHolder holder = null;
-		try {
-			if (KafkaUtils.MICROMETER_PRESENT) {
-				holder = new MicrometerHolder(this.applicationContext, this.beanName,
-						"spring.kafka.template", "KafkaTemplate Timer",
-						this.micrometerTags);
-			}
-		}
-		catch (@SuppressWarnings("unused") IllegalStateException ex) {
-			this.micrometerEnabled = false;
-		}
-		return holder;
-	}
-
 	@Override
 	public void destroy() {
-		if (this.micrometerHolder != null) {
-			this.micrometerHolder.destroy();
-		}
 		if (this.customProducerFactory) {
 			((DefaultKafkaProducerFactory<K, V>) this.producerFactory).destroy();
 		}

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

@@ -260,8 +260,6 @@ public enum EOSMode {
 
 	private long idleBetweenPolls;
 
-	private boolean micrometerEnabled = true;
-
 	private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT;
 
 	private Boolean subBatchPerPartition;
@@ -622,17 +620,19 @@ public long getIdleBetweenPolls() {
 		return this.idleBetweenPolls;
 	}
 
+	@Deprecated
 	public boolean isMicrometerEnabled() {
-		return this.micrometerEnabled;
+		return true;
 	}
 
 	/**
 	 * Set to false to disable the Micrometer listener timers. Default true.
 	 * @param micrometerEnabled false to disable.
 	 * @since 2.3
+	 * @deprecated this is no longer used.
 	 */
+	@Deprecated
 	public void setMicrometerEnabled(boolean micrometerEnabled) {
-		this.micrometerEnabled = micrometerEnabled;
 	}
 
 	/**