Add observationEnabled properties for Apache Kafka

See gh-38057

Author: Zhiyang.Wang1

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java

@@ -236,6 +236,7 @@ private void configureContainer(ContainerProperties container) {
 		map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
 		map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
 		map.from(properties::isImmediateStop).to(container::setStopImmediate);
+		map.from(properties::getObservationEnabled).to(container::setObservationEnabled);
 		map.from(this.transactionManager).to(container::setTransactionManager);
 		map.from(this.rebalanceListener).to(container::setConsumerRebalanceListener);
 		map.from(this.listenerTaskExecutor).to(container::setListenerTaskExecutor);

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

@@ -98,6 +98,7 @@ PropertiesKafkaConnectionDetails kafkaConnectionDetails(KafkaProperties properti
 		map.from(kafkaProducerListener).to(kafkaTemplate::setProducerListener);
 		map.from(this.properties.getTemplate().getDefaultTopic()).to(kafkaTemplate::setDefaultTopic);
 		map.from(this.properties.getTemplate().getTransactionIdPrefix()).to(kafkaTemplate::setTransactionIdPrefix);
+		map.from(this.properties.getTemplate().getObservationEnabled()).to(kafkaTemplate::setObservationEnabled);
 		return kafkaTemplate;
 	}
 

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java

@@ -984,6 +984,11 @@ public static class Template {
 		 */
 		private String transactionIdPrefix;
 
+		/**
+		 * Whether to enable observation.
+		 */
+		private boolean observationEnabled;
+
 		public String getDefaultTopic() {
 			return this.defaultTopic;
 		}
@@ -1000,6 +1005,14 @@ public void setTransactionIdPrefix(String transactionIdPrefix) {
 			this.transactionIdPrefix = transactionIdPrefix;
 		}
 
+		public boolean getObservationEnabled() {
+			return this.observationEnabled;
+		}
+
+		public void setObservationEnabled(boolean observationEnabled) {
+			this.observationEnabled = observationEnabled;
+		}
+
 	}
 
 	public static class Listener {
@@ -1117,6 +1130,11 @@ public enum Type {
 		 */
 		private Boolean changeConsumerThreadName;
 
+		/**
+		 * Whether to enable observation.
+		 */
+		private boolean observationEnabled;
+
 		public Type getType() {
 			return this.type;
 		}
@@ -1261,6 +1279,14 @@ public void setChangeConsumerThreadName(Boolean changeConsumerThreadName) {
 			this.changeConsumerThreadName = changeConsumerThreadName;
 		}
 
+		public boolean getObservationEnabled() {
+			return this.observationEnabled;
+		}
+
+		public void setObservationEnabled(boolean observationEnabled) {
+			this.observationEnabled = observationEnabled;
+		}
+
 	}
 
 	public static class Ssl {

spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -1811,6 +1811,10 @@
       "name": "spring.kafka.jaas.control-flag",
       "defaultValue": "required"
     },
+    {
+      "name": "spring.kafka.listener.observation-enabled",
+      "defaultValue": false
+    },
     {
       "name": "spring.kafka.listener.only-log-record-metadata",
       "type": "java.lang.Boolean",
@@ -1905,6 +1909,10 @@
         "level": "error"
       }
     },
+    {
+      "name": "spring.kafka.template.observation-enabled",
+      "defaultValue": false
+    },
     {
       "name": "spring.liquibase.check-change-log-location",
       "type": "java.lang.Boolean",

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

@@ -618,7 +618,8 @@ void listenerProperties() {
 					"spring.kafka.listener.missing-topics-fatal=true", "spring.kafka.jaas.enabled=true",
 					"spring.kafka.listener.immediate-stop=true", "spring.kafka.producer.transaction-id-prefix=foo",
 					"spring.kafka.jaas.login-module=foo", "spring.kafka.jaas.control-flag=REQUISITE",
-					"spring.kafka.jaas.options.useKeyTab=true", "spring.kafka.listener.async-acks=true")
+					"spring.kafka.jaas.options.useKeyTab=true", "spring.kafka.listener.async-acks=true",
+					"spring.kafka.template.observation-enabled=true", "spring.kafka.listener.observation-enabled=true")
 			.run((context) -> {
 				DefaultKafkaProducerFactory<?, ?> producerFactory = context.getBean(DefaultKafkaProducerFactory.class);
 				DefaultKafkaConsumerFactory<?, ?> consumerFactory = context.getBean(DefaultKafkaConsumerFactory.class);
@@ -629,6 +630,7 @@ void listenerProperties() {
 				assertThat(kafkaTemplate).hasFieldOrPropertyWithValue("producerFactory", producerFactory);
 				assertThat(kafkaTemplate.getDefaultTopic()).isEqualTo("testTopic");
 				assertThat(kafkaTemplate).hasFieldOrPropertyWithValue("transactionIdPrefix", "txOverride");
+				assertThat(kafkaTemplate).hasFieldOrPropertyWithValue("observationEnabled", true);
 				assertThat(kafkaListenerContainerFactory.getConsumerFactory()).isEqualTo(consumerFactory);
 				ContainerProperties containerProperties = kafkaListenerContainerFactory.getContainerProperties();
 				assertThat(containerProperties.getAckMode()).isEqualTo(AckMode.MANUAL);
@@ -645,6 +647,7 @@ void listenerProperties() {
 				assertThat(containerProperties.isLogContainerConfig()).isTrue();
 				assertThat(containerProperties.isMissingTopicsFatal()).isTrue();
 				assertThat(containerProperties.isStopImmediate()).isTrue();
+				assertThat(containerProperties.isObservationEnabled()).isTrue();
 				assertThat(kafkaListenerContainerFactory).extracting("concurrency").isEqualTo(3);
 				assertThat(kafkaListenerContainerFactory.isBatchListener()).isTrue();
 				assertThat(kafkaListenerContainerFactory).hasFieldOrPropertyWithValue("autoStartup", true);