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": [