From 1b8f955fec5713eb13031274ca1f217c2f1b2bd3 Mon Sep 17 00:00:00 2001
From: Ryan Dunckel <Ryan.Dunckel@fisglobal.com>
Date: Mon, 1 Jul 2019 14:22:42 -0500
Subject: [PATCH] Add config property for Kafka consumer isolation level

See gh-17389
---
 .../autoconfigure/kafka/KafkaProperties.java   | 16 ++++++++++++++++
 ...ditional-spring-configuration-metadata.json | 18 ++++++++++++++++++
 2 files changed, 34 insertions(+)

diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
index 0715989686..f7d80cf4b8 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaProperties.java
@@ -266,6 +266,13 @@ public class KafkaProperties {
 		 */
 		private Duration heartbeatInterval;
 
+		/**
+		 * Controls how transactional messages are returned when polling the broker
+		 * (non-transactional messages will be unconditionally returned, regardless of
+		 * this setting).
+		 */
+		private String isolationLevel;
+
 		/**
 		 * Deserializer class for keys.
 		 */
@@ -362,6 +369,14 @@ public class KafkaProperties {
 			this.heartbeatInterval = heartbeatInterval;
 		}
 
+		public String getIsolationLevel() {
+			return this.isolationLevel;
+		}
+
+		public void setIsolationLevel(String isolationLevel) {
+			this.isolationLevel = isolationLevel;
+		}
+
 		public Class<?> getKeyDeserializer() {
 			return this.keyDeserializer;
 		}
@@ -406,6 +421,7 @@ public class KafkaProperties {
 			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(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));
diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json
index 8eda844d5b..f8d378ee39 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json
@@ -2228,6 +2228,24 @@
         }
       ]
     },
+    {
+      "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": [