Handle errors for non-existing topics (#514)

* Handle errors for non-existing topics

When topic creation is disabled both on the binder and the broker,
the binder currently throws an NPE. Catching this situation and
throw a more graceful error to the user.

Adding tests to verify.

Resolves #513

* Addressing PR review comments

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -40,6 +40,7 @@ import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.errors.TopicExistsException;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -60,6 +61,7 @@ import org.springframework.retry.backoff.ExponentialBackOffPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
@@ -277,12 +279,14 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		try {
 			createTopicIfNecessary(adminClient, name, partitionCount, tolerateLowerPartitionsOnBroker, properties);
 		}
+		//TODO: Remove catching Throwable. See this thread: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/514#discussion_r241075940
 		catch (Throwable throwable) {
 			if (throwable instanceof Error) {
 				throw (Error) throwable;
 			}
 			else {
-				throw new ProvisioningException("provisioning exception", throwable);
+				//TODO: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/514#discussion_r241075940
+				throw new ProvisioningException("Provisioning exception", throwable);
 			}
 		}
 	}
@@ -393,11 +397,38 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 
 	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount,
 														final boolean tolerateLowerPartitionsOnBroker,
-														final Callable<Collection<PartitionInfo>> callable) {
+														final Callable<Collection<PartitionInfo>> callable,
+														final String topicName) {
 		try {
 			return this.metadataRetryOperations
 					.execute((context) -> {
-						Collection<PartitionInfo> partitions = callable.call();
+						Collection<PartitionInfo> partitions = Collections.emptyList();
+
+						try {
+							//This call may return null or throw an exception.
+							partitions = callable.call();
+						}
+						catch (Exception ex) {
+							//The above call can potentially throw exceptions such as timeout. If we can determine
+							//that the exception was due to an unknown topic on the broker, just simply rethrow that.
+							if (ex instanceof UnknownTopicOrPartitionException) {
+								throw ex;
+							}
+						}
+						if (CollectionUtils.isEmpty(partitions)) {
+							final AdminClient adminClient = AdminClient.create(this.adminClientProperties);
+							final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Collections.singletonList(topicName));
+							try {
+								describeTopicsResult.all().get();
+							}
+							catch (ExecutionException ex) {
+								if (ex.getCause() instanceof UnknownTopicOrPartitionException) {
+									throw new Exception(ex.getCause());
+								} else {
+									logger.warn("No partitions have been retrieved for the topic (" + topicName + "). This will affect the health check.");
+								}
+							}
+						}
 						// do a sanity check on the partition set
 						int partitionSize = partitions.size();
 						if (partitionSize < partitionCount) {

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -289,7 +289,7 @@ public class KafkaMessageChannelBinder extends
 					producer.close();
 					((DisposableBean) producerFB).destroy();
 					return partitionsFor;
-				});
+				}, destination.getName());
 		this.topicsInUse.put(destination.getName(), new TopicInformation(null, partitions, false));
 		if (producerProperties.isPartitioned() && producerProperties.getPartitionCount() < partitions.size()) {
 			if (this.logger.isInfoEnabled()) {
@@ -744,7 +744,7 @@ public class KafkaMessageChannelBinder extends
 						List<PartitionInfo> partitionsFor = consumer.partitionsFor(topic);
 						return partitionsFor;
 					}
-				});
+				}, topic);
 	}
 
 	@Override

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java

@@ -0,0 +1,113 @@
+/*
+ * 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.cloud.stream.binder.kafka;
+
+import java.util.Collections;
+
+import kafka.server.KafkaConfig;
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+import org.springframework.cloud.stream.binder.BinderException;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.retry.policy.SimpleRetryPolicy;
+import org.springframework.retry.support.RetryTemplate;
+
+import static org.hamcrest.CoreMatchers.isA;
+
+/**
+ * @author Soby Chacko
+ */
+public class AutoCreateTopicDisabledTests {
+
+	@Rule
+	public ExpectedException expectedException = ExpectedException.none();
+
+	@ClassRule
+	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 1)
+			.brokerProperty(KafkaConfig.AutoCreateTopicsEnableProp(), "false");
+
+	@Test
+	public void testAutoCreateTopicDisabledFailsOnConsumerIfTopicNonExistentOnBroker() throws Throwable {
+
+		KafkaProperties kafkaProperties = new TestKafkaProperties();
+		kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getEmbeddedKafka().getBrokersAsString()));
+		KafkaBinderConfigurationProperties configurationProperties =
+				new KafkaBinderConfigurationProperties(kafkaProperties);
+		//disable auto create topic on the binder.
+		configurationProperties.setAutoCreateTopics(false);
+
+		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties);
+		provisioningProvider.setMetadataRetryOperations(new RetryTemplate());
+
+		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties, provisioningProvider);
+
+		final String testTopicName = "nonExistent" + System.currentTimeMillis();
+
+		ExtendedConsumerProperties<KafkaConsumerProperties> properties = new ExtendedConsumerProperties<>(new KafkaConsumerProperties());
+
+		expectedException.expect(BinderException.class);
+		expectedException
+				.expectCause(isA(UnknownTopicOrPartitionException.class));
+		binder.createConsumerEndpoint(() -> testTopicName, "group", properties);
+	}
+
+
+	@Test
+	public void testAutoCreateTopicDisabledFailsOnProducerIfTopicNonExistentOnBroker() throws Throwable {
+
+		KafkaProperties kafkaProperties = new TestKafkaProperties();
+		kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getEmbeddedKafka().getBrokersAsString()));
+
+		KafkaBinderConfigurationProperties configurationProperties =
+				new KafkaBinderConfigurationProperties(kafkaProperties);
+		//disable auto create topic on the binder.
+		configurationProperties.setAutoCreateTopics(false);
+		//reduce the wait time on the producer blocking operations.
+		configurationProperties.getConfiguration().put("max.block.ms", "3000");
+
+		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties);
+		SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(1);
+		final RetryTemplate metadataRetryOperations = new RetryTemplate();
+		metadataRetryOperations.setRetryPolicy(simpleRetryPolicy);
+		provisioningProvider.setMetadataRetryOperations(metadataRetryOperations);
+
+		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(configurationProperties, provisioningProvider);
+
+		final String testTopicName = "nonExistent" + System.currentTimeMillis();
+
+		ExtendedProducerProperties<KafkaProducerProperties> properties = new ExtendedProducerProperties<>(new KafkaProducerProperties());
+
+		expectedException.expect(BinderException.class);
+		expectedException
+				.expectCause(isA(UnknownTopicOrPartitionException.class));
+
+		binder.bindProducer(testTopicName, new DirectChannel(), properties);
+
+	}
+}

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java

@@ -167,7 +167,7 @@ public class KafkaBinderUnitTests {
 			return partitions.stream()
 					.map(p -> new PartitionInfo(topic, part.getAndIncrement(), null, null, null))
 					.collect(Collectors.toList());
-		}).given(provisioningProvider).getPartitionsForTopic(anyInt(), anyBoolean(), any());
+		}).given(provisioningProvider).getPartitionsForTopic(anyInt(), anyBoolean(), any(), any());
 		@SuppressWarnings("unchecked")
 		final Consumer<byte[], byte[]> consumer = mock(Consumer.class);
 		final CountDownLatch latch = new CountDownLatch(1);