Add missingTopicsFatal to container factory

See spring-projects/spring-boot#16727

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -96,6 +96,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private ReplyHeadersConfigurer replyHeadersConfigurer;
 
+	private Boolean missingTopicsFatal;
+
 	/**
 	 * Specify a {@link ConsumerFactory} to use.
 	 * @param consumerFactory The consumer factory.
@@ -258,6 +260,17 @@ public void setReplyHeadersConfigurer(ReplyHeadersConfigurer replyHeadersConfigu
 		this.replyHeadersConfigurer = replyHeadersConfigurer;
 	}
 
+	/**
+	 * Set to false to allow the container to start even if any of the configured topics
+	 * are not present on the broker. Does not apply when topic patterns are configured.
+	 * Default true;
+	 * @param missingTopicsFatal the missingTopicsFatal.
+	 * @since 2.3
+	 */
+	public void setMissingTopicsFatal(boolean missingTopicsFatal) {
+		this.missingTopicsFatal = missingTopicsFatal;
+	}
+
 	/**
 	 * Obtain the properties template for this factory - set properties as needed
 	 * and they will be copied to a final properties instance for the endpoint.
@@ -335,7 +348,8 @@ protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 					properties::setAckCount)
 			.acceptIfCondition(this.containerProperties.getAckTime() > 0, this.containerProperties.getAckTime(),
 					properties::setAckTime)
-			.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler);
+			.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
+			.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
 		if (endpoint.getAutoStartup() != null) {
 			instance.setAutoStartup(endpoint.getAutoStartup());
 		}

spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java

@@ -62,6 +62,7 @@
 import org.springframework.kafka.annotation.KafkaHandler;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -131,6 +132,9 @@ public class ReplyingKafkaTemplateTests {
 	@Autowired
 	private Config config;
 
+	@Autowired
+	private KafkaListenerEndpointRegistry registry;
+
 	@Test
 	public void testGood() throws Exception {
 		ReplyingKafkaTemplate<Integer, String, String> template = createTemplate(A_REPLY);
@@ -147,6 +151,10 @@ public void testGood() throws Exception {
 			new DefaultKafkaHeaderMapper().toHeaders(consumerRecord.headers(), receivedHeaders);
 			assertThat(receivedHeaders).containsKey("baz");
 			assertThat(receivedHeaders).hasSize(2);
+			assertThat(KafkaTestUtils.getPropertyValue(
+					this.registry.getListenerContainer(A_REQUEST), "containerProperties.missingTopicsFatal",
+					Boolean.class))
+				.isFalse();
 		}
 		finally {
 			template.stop();
@@ -501,6 +509,7 @@ public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerCon
 			factory.setConsumerFactory(cf());
 			factory.setReplyTemplate(template());
 			factory.setReplyHeadersConfigurer((k, v) -> k.equals("baz"));
+			factory.setMissingTopicsFatal(false);
 			return factory;
 		}
 
@@ -529,7 +538,7 @@ public Map<String, Object> additionalHeaders() {
 			return factory;
 		}
 
-		@KafkaListener(topics = A_REQUEST)
+		@KafkaListener(id = A_REQUEST, topics = A_REQUEST)
 		@SendTo  // default REPLY_TOPIC header
 		public String handleA(String in) {
 			return in.toUpperCase();