GH-696: Add missingTopicsFatal container property

Fixes #696

Container start() fails if topics are missing.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -16,12 +16,16 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
@@ -181,6 +185,7 @@ public boolean isAutoStartup() {
 		return this.autoStartup;
 	}
 
+	@Override
 	public void setAutoStartup(boolean autoStartup) {
 		this.autoStartup = autoStartup;
 	}
@@ -251,6 +256,33 @@ public final void start() {
 		}
 	}
 
+	protected void checkTopics() {
+		if (this.containerProperties.isMissingTopicsFatal() && this.containerProperties.getTopicPattern() == null) {
+			try (Consumer<K, V> consumer = this.consumerFactory.createConsumer(this.containerProperties.getGroupId(),
+					this.containerProperties.getClientId(), null)) {
+				if (consumer != null) {
+					String[] topics = this.containerProperties.getTopics();
+					if (topics == null) {
+						topics = Arrays.stream(this.containerProperties.getTopicPartitions())
+								.map(tp -> tp.topic())
+								.toArray(String[]::new);
+					}
+					List<String> missing = new ArrayList<>();
+					for (String topic : topics) {
+						if (consumer.partitionsFor(topic) == null) {
+							missing.add(topic);
+						}
+					}
+					if (missing.size() > 0) {
+						throw new IllegalStateException(
+								"Topic(s) " + missing.toString()
+										+ " is/are not present and missingTopicsFatal is true");
+					}
+				}
+			}
+		}
+	}
+
 	public void checkGroupId() {
 		if (this.containerProperties.getTopicPartitions() == null) {
 			boolean hasGroupIdConsumerConfig = true; // assume true for non-standard containers

spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java

@@ -132,6 +132,7 @@ public boolean isContainerPaused() {
 	@Override
 	protected void doStart() {
 		if (!isRunning()) {
+			checkTopics();
 			ContainerProperties containerProperties = getContainerProperties();
 			TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();
 			if (topicPartitions != null

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

@@ -206,6 +206,8 @@ public enum AckMode {
 
 	private LogIfLevelEnabled.Level commitLogLevel = LogIfLevelEnabled.Level.DEBUG;
 
+	private boolean missingTopicsFatal = true;
+
 	public ContainerProperties(String... topics) {
 		Assert.notEmpty(topics, "An array of topicPartitions must be provided");
 		this.topics = Arrays.asList(topics).toArray(new String[topics.length]);
@@ -556,6 +558,27 @@ public void setCommitLogLevel(LogIfLevelEnabled.Level commitLogLevel) {
 		this.commitLogLevel = commitLogLevel;
 	}
 
+	/**
+	 * If true, the container won't start if any of the configured topics are not present
+	 * on the broker. Does not apply when topic patterns are configured. Default true;
+	 * @return the missingTopicsFatal.
+	 * @since 2.2
+	 */
+	public boolean isMissingTopicsFatal() {
+		return this.missingTopicsFatal;
+	}
+
+	/**
+	 * 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.2
+	 */
+	public void setMissingTopicsFatal(boolean missingTopicsFatal) {
+		this.missingTopicsFatal = missingTopicsFatal;
+	}
+
 	@Override
 	public String toString() {
 		return "ContainerProperties ["

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -227,6 +227,9 @@ protected void doStart() {
 		if (isRunning()) {
 			return;
 		}
+		if (this.clientIdSuffix == null) { // stand-alone container
+			checkTopics();
+		}
 		ContainerProperties containerProperties = getContainerProperties();
 		if (!this.consumerFactory.isAutoCommit()) {
 			AckMode ackMode = containerProperties.getAckMode();

spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
+
+import java.util.Map;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.test.rule.KafkaEmbedded;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class MissingTopicsTests {
+
+	@ClassRule
+	public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true)
+		.brokerProperty("auto.create.topics.enable", false);
+
+	@Test
+	public void testMissingTopicCMLC() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps("missing1", "true", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties("notexisting");
+		containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
+		ConcurrentMessageListenerContainer<Integer, String> container =
+				new ConcurrentMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testMissing1");
+		try {
+			container.start();
+			fail("Expected exception");
+		}
+		catch (IllegalStateException e) {
+			assertThat(e.getMessage()).contains("missingTopicsFatal");
+		}
+	}
+
+	@Test
+	public void testMissingTopicKMLC() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps("missing2", "true", embeddedKafka);
+		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
+		ContainerProperties containerProps = new ContainerProperties("notexisting");
+		containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
+		KafkaMessageListenerContainer<Integer, String> container =
+				new KafkaMessageListenerContainer<>(cf, containerProps);
+		container.setBeanName("testMissing2");
+		try {
+			container.start();
+			fail("Expected exception");
+		}
+		catch (IllegalStateException e) {
+			assertThat(e.getMessage()).contains("missingTopicsFatal");
+		}
+		container.getContainerProperties().setMissingTopicsFatal(false);
+		container.start();
+		container.stop();
+	}
+
+
+}

src/reference/asciidoc/kafka.adoc

@@ -616,12 +616,17 @@ return container;
 
 Refer to the JavaDocs for `ContainerProperties` for more information about the various properties that can be set.
 
-Since version _2.1.1_, a new property `logContainerConfig` is available; when true, and INFO logging is enabled, each listener container will write a log message summarizing its configuration properties.
+Since _version 2.1.1_, a new property `logContainerConfig` is available; when true, and INFO logging is enabled, each listener container will write a log message summarizing its configuration properties.
 
 By default, logging of topic offset commits is performed with the DEBUG logging level.
 Starting with _version 2.1.2_, there is a new property in `ContainerProperties` called `commitLogLevel` which allows you to specify the log level for these messages.
 For example, to change the log level to INFO, use `containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO);`.
 
+Starting with _version 2.2_, a new container property `missingTopicsFatal` has been added (default `true`).
+This prevents the container from starting if any of the configured topics are not present on the broker; it does not apply if the container is configured to listen to a topic pattern (regex).
+Previously, the container threads looped within the `consumer.poll()` method waiting for the topic to appear, while logging many messages; aside from the logs, there was no indication that there was a problem.
+To restore the previous behavior, set the property to `false`.
+
 ====== ConcurrentMessageListenerContainer
 
 The single constructor is similar to the first `KafkaListenerContainer` constructor:

src/reference/asciidoc/whats-new.adoc

@@ -20,3 +20,9 @@ A new `AfterRollbackProcessor` strategy is provided - see <<after-rollback>> for
 
 The `ConcurrentKafkaListenerContainerFactory` can now be used to create/configure any `ConcurrentMessageListenerContainer`, not just those for `@KafkaListener` annotations.
 See <<container-factory>> for more information.
+
+==== Listener Container Changes
+
+A new container property `missingTopicsFatal` has been added.
+
+See <<kafka-container>> for more information.