Polish "Add config property for Kafka consumer isolation level"

See gh-17389

Author: wilkinsona

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

@@ -23,12 +23,14 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.requests.IsolationLevel;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 
@@ -267,11 +269,9 @@ public static class Consumer {
 		private Duration heartbeatInterval;
 
 		/**
-		 * Controls how transactional messages are returned when polling the broker
-		 * (non-transactional messages will be unconditionally returned, regardless of
-		 * this setting).
+		 * Isolation level for reading messages that have been written transactionally.
 		 */
-		private String isolationLevel;
+		private IsolationLevel isolationLevel = IsolationLevel.READ_UNCOMMITTED;
 
 		/**
 		 * Deserializer class for keys.
@@ -369,11 +369,11 @@ public void setHeartbeatInterval(Duration heartbeatInterval) {
 			this.heartbeatInterval = heartbeatInterval;
 		}
 
-		public String getIsolationLevel() {
+		public IsolationLevel getIsolationLevel() {
 			return this.isolationLevel;
 		}
 
-		public void setIsolationLevel(String isolationLevel) {
+		public void setIsolationLevel(IsolationLevel isolationLevel) {
 			this.isolationLevel = isolationLevel;
 		}
 
@@ -421,7 +421,8 @@ public Map<String, Object> buildProperties() {
 			map.from(this::getGroupId).to(properties.in(ConsumerConfig.GROUP_ID_CONFIG));
 			map.from(this::getHeartbeatInterval).asInt(Duration::toMillis)
 					.to(properties.in(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG));
-			map.from(this::getIsolationLevel).to(properties.in(ConsumerConfig.ISOLATION_LEVEL_CONFIG));
+			map.from(() -> getIsolationLevel().name().toLowerCase(Locale.ROOT))
+					.to(properties.in(ConsumerConfig.ISOLATION_LEVEL_CONFIG));
 			map.from(this::getKeyDeserializer).to(properties.in(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
 			map.from(this::getValueDeserializer).to(properties.in(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));
 			map.from(this::getMaxPollRecords).to(properties.in(ConsumerConfig.MAX_POLL_RECORDS_CONFIG));

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

@@ -2228,24 +2228,6 @@
         }
       ]
     },
-    {
-      "name": "spring.kafka.consumer.isolation-level",
-      "values": [
-        {
-          "value": "read_committed",
-          "description": "Only consume transactional messages that have been committed."
-        },
-        {
-          "value": "read_uncommitted",
-          "description": "Consume all transactional messages (even those that have been aborted)."
-        }
-      ],
-      "providers": [
-        {
-          "name": "any"
-        }
-      ]
-    },
     {
       "name": "spring.kafka.producer.key-serializer",
       "providers": [

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

@@ -105,6 +105,7 @@ void consumerProperties() {
 				"spring.kafka.consumer.enable-auto-commit=false", "spring.kafka.consumer.fetch-max-wait=456",
 				"spring.kafka.consumer.properties.fiz.buz=fix.fox", "spring.kafka.consumer.fetch-min-size=1KB",
 				"spring.kafka.consumer.group-id=bar", "spring.kafka.consumer.heartbeat-interval=234",
+				"spring.kafka.consumer.isolation-level = read-committed",
 				"spring.kafka.consumer.key-deserializer = org.apache.kafka.common.serialization.LongDeserializer",
 				"spring.kafka.consumer.value-deserializer = org.apache.kafka.common.serialization.IntegerDeserializer")
 				.run((context) -> {
@@ -133,6 +134,7 @@ void consumerProperties() {
 					assertThat(configs.get(ConsumerConfig.FETCH_MIN_BYTES_CONFIG)).isEqualTo(1024);
 					assertThat(configs.get(ConsumerConfig.GROUP_ID_CONFIG)).isEqualTo("bar");
 					assertThat(configs.get(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG)).isEqualTo(234);
+					assertThat(configs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG)).isEqualTo("read_committed");
 					assertThat(configs.get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG))
 							.isEqualTo(LongDeserializer.class);
 					assertThat(configs.get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG))