From 3c46b9e83de45f262f77fe6a59e3d04efff17c02 Mon Sep 17 00:00:00 2001
From: Stephane Nicoll <snicoll@pivotal.io>
Date: Mon, 20 May 2019 14:45:59 +0200
Subject: [PATCH] Polish "Add configuration property for Spring Kafka's
 missingTopicsFatal"

Closes gh-16740
---
 ...entKafkaListenerContainerFactoryConfigurer.java |  2 +-
 .../boot/autoconfigure/kafka/KafkaProperties.java  |  9 +++++----
 .../kafka/KafkaAutoConfigurationTests.java         | 14 ++++++++++++++
 3 files changed, 20 insertions(+), 5 deletions(-)

diff --git a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
index 232a7e51b41..1631f5980d9 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/ConcurrentKafkaListenerContainerFactoryConfigurer.java
@@ -158,8 +158,8 @@ public class ConcurrentKafkaListenerContainerFactoryConfigurer {
 		map.from(properties::getMonitorInterval).as(Duration::getSeconds)
 				.as(Number::intValue).to(container::setMonitorInterval);
 		map.from(properties::getLogContainerConfig).to(container::setLogContainerConfig);
+		map.from(properties::isMissingTopicsFatal).to(container::setMissingTopicsFatal);
 		map.from(this.transactionManager).to(container::setTransactionManager);
-		map.from(properties::getMissingTopicsFatal).to(container::setMissingTopicsFatal);
 	}
 
 }
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 e82c88e6d3b..1d51bc0c114 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
@@ -874,9 +874,10 @@ public class KafkaProperties {
 		private Boolean logContainerConfig;
 
 		/**
-		 * Set to false to disable checking that topic(s) exist.
+		 * Whether the container should fail to start if at least one of the configured
+		 * topics are not present on the broker.
 		 */
-		private Boolean missingTopicsFatal;
+		private boolean missingTopicsFatal = true;
 
 		public Type getType() {
 			return this.type;
@@ -966,11 +967,11 @@ public class KafkaProperties {
 			this.logContainerConfig = logContainerConfig;
 		}
 
-		public Boolean getMissingTopicsFatal() {
+		public boolean isMissingTopicsFatal() {
 			return this.missingTopicsFatal;
 		}
 
-		public void setMissingTopicsFatal(Boolean missingTopicsFatal) {
+		public void setMissingTopicsFatal(boolean missingTopicsFatal) {
 			this.missingTopicsFatal = missingTopicsFatal;
 		}
 
diff --git a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
index 444efdc2de9..935af69698e 100644
--- a/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
+++ b/spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
@@ -38,6 +38,7 @@ import org.apache.kafka.streams.StreamsConfig;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Listener;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -511,6 +512,19 @@ public class KafkaAutoConfigurationTests {
 				});
 	}
 
+	@Test
+	public void listenerPropertiesMatchDefaults() {
+		this.contextRunner.run((context) -> {
+			Listener listenerProperties = new KafkaProperties().getListener();
+			AbstractKafkaListenerContainerFactory<?, ?, ?> kafkaListenerContainerFactory = (AbstractKafkaListenerContainerFactory<?, ?, ?>) context
+					.getBean(KafkaListenerContainerFactory.class);
+			ContainerProperties containerProperties = kafkaListenerContainerFactory
+					.getContainerProperties();
+			assertThat(containerProperties.isMissingTopicsFatal())
+					.isEqualTo(listenerProperties.isMissingTopicsFatal());
+		});
+	}
+
 	@Test
 	public void testKafkaTemplateRecordMessageConverters() {
 		this.contextRunner.withUserConfiguration(MessageConverterConfiguration.class)