diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index d4e1d0396..9b8932950 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/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 getPartitionsForTopic(final int partitionCount, final boolean tolerateLowerPartitionsOnBroker, - final Callable> callable) { + final Callable> callable, + final String topicName) { try { return this.metadataRetryOperations .execute((context) -> { - Collection partitions = callable.call(); + Collection 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) { diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index fab7adc06..b153293bc 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 partitionsFor = consumer.partitionsFor(topic); return partitionsFor; } - }); + }, topic); } @Override diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java new file mode 100644 index 000000000..8537c557e --- /dev/null +++ b/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 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 properties = new ExtendedProducerProperties<>(new KafkaProducerProperties()); + + expectedException.expect(BinderException.class); + expectedException + .expectCause(isA(UnknownTopicOrPartitionException.class)); + + binder.bindProducer(testTopicName, new DirectChannel(), properties); + + } +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java index 82627cd85..6dcf4862e 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java +++ b/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 consumer = mock(Consumer.class); final CountDownLatch latch = new CountDownLatch(1);