From 8d5bf66c7f79802cccc0fdc6218c13ac517b695b Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Sun, 1 May 2016 17:29:43 -0400 Subject: [PATCH] GH-504: AutoConfiguration of Kafka Topics Add a flag for controlling topic autoconfiguration by the Kafka binder Fixes #504 - introduces a new binder setting `autoConfigureTopics` that allows the user to disable the automatic creation of topics by the binder Additional tests for partitioning Adding missing documentation for `replicationFactor` --- .../kafka/KafkaMessageChannelBinder.java | 69 +++++--- .../config/KafkaBinderConfiguration.java | 2 +- .../KafkaBinderConfigurationProperties.java | 9 + .../stream/binder/kafka/KafkaBinderTests.java | 167 ++++++++++++++++++ .../spring-cloud-stream-overview.adoc | 35 ++-- .../test/junit/kafka/KafkaTestSupport.java | 17 +- 6 files changed, 251 insertions(+), 48 deletions(-) diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index c38c6f274..375f0d028 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -32,6 +32,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicInteger; +import kafka.admin.AdminUtils; +import kafka.api.OffsetRequest; +import kafka.serializer.Decoder; +import kafka.serializer.DefaultDecoder; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; @@ -96,12 +102,6 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; -import kafka.admin.AdminUtils; -import kafka.api.OffsetRequest; -import kafka.serializer.Decoder; -import kafka.serializer.DefaultDecoder; -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils; import scala.collection.Seq; /** @@ -126,7 +126,9 @@ public class KafkaMessageChannelBinder public static final DaemonThreadFactory DAEMON_THREAD_FACTORY = new DaemonThreadFactory(); - private RetryOperations retryOperations; + private boolean autoConfigureTopics = true; + + private RetryOperations metadataRetryOperations; private final Map> topicsInUse = new HashMap<>(); @@ -219,10 +221,10 @@ public class KafkaMessageChannelBinder /** * Retry configuration for operations such as validating topic creation - * @param retryOperations the retry configuration + * @param metadataRetryOperations the retry configuration */ - public void setRetryOperations(RetryOperations retryOperations) { - this.retryOperations = retryOperations; + public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) { + this.metadataRetryOperations = metadataRetryOperations; } public void setExtendedBindingProperties(KafkaExtendedBindingProperties extendedBindingProperties) { @@ -237,7 +239,7 @@ public class KafkaMessageChannelBinder DefaultConnectionFactory defaultConnectionFactory = new DefaultConnectionFactory(configuration); defaultConnectionFactory.afterPropertiesSet(); this.connectionFactory = defaultConnectionFactory; - if (retryOperations == null) { + if (metadataRetryOperations == null) { RetryTemplate retryTemplate = new RetryTemplate(); SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(); @@ -249,7 +251,7 @@ public class KafkaMessageChannelBinder backOffPolicy.setMultiplier(2); backOffPolicy.setMaxInterval(1000); retryTemplate.setBackOffPolicy(backOffPolicy); - retryOperations = retryTemplate; + metadataRetryOperations = retryTemplate; } } @@ -320,6 +322,14 @@ public class KafkaMessageChannelBinder this.zkConnectionTimeout = zkConnectionTimeout; } + public boolean isAutoConfigureTopics() { + return autoConfigureTopics; + } + + public void setAutoConfigureTopics(boolean autoConfigureTopics) { + this.autoConfigureTopics = autoConfigureTopics; + } + @Override public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) { return extendedBindingProperties.getExtendedConsumerProperties(channelName); @@ -419,7 +429,7 @@ public class KafkaMessageChannelBinder * Creates a Kafka topic if needed, or try to increase its partition count to the * desired number. */ - private Collection ensureTopicCreated(final String topicName, final int numPartitions, + private Collection ensureTopicCreated(final String topicName, final int minExpectedPartitions, int replicationFactor) { final ZkClient zkClient = new ZkClient(zkAddress, getZkSessionTimeout(), getZkConnectionTimeout(), @@ -428,30 +438,33 @@ public class KafkaMessageChannelBinder // The following is basically copy/paste from AdminUtils.createTopic() with // createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true) final Properties topicConfig = new Properties(); - Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); - final scala.collection.Map> replicaAssignment = AdminUtils - .assignReplicasToBrokers(brokerList, numPartitions, replicationFactor, -1, -1); - retryOperations.execute(new RetryCallback() { + if (isAutoConfigureTopics()) { + Seq brokerList = ZkUtils.getSortedBrokerList(zkClient); + final scala.collection.Map> replicaAssignment = AdminUtils + .assignReplicasToBrokers(brokerList, minExpectedPartitions, replicationFactor, -1, -1); + metadataRetryOperations.execute(new RetryCallback() { - @Override - public Object doWithRetry(RetryContext context) throws RuntimeException { - AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, replicaAssignment, - topicConfig, true); - return null; - } - }); + @Override + public Object doWithRetry(RetryContext context) throws RuntimeException { + AdminUtils.createOrUpdateTopicPartitionAssignmentPathInZK(zkClient, topicName, replicaAssignment, + topicConfig, true); + return null; + } + }); + } try { - Collection partitions = retryOperations + Collection partitions = metadataRetryOperations .execute(new RetryCallback, Exception>() { @Override public Collection doWithRetry(RetryContext context) throws Exception { connectionFactory.refreshMetadata(Collections.singleton(topicName)); Collection partitions = connectionFactory.getPartitions(topicName); - if (partitions.size() < numPartitions) { + if (partitions.size() < minExpectedPartitions) { throw new IllegalStateException( - "The number of expected partitions was: " + numPartitions + ", but " - + partitions.size() + " have been found instead"); + "The number of expected partitions was: " + minExpectedPartitions + ", but " + + partitions.size() + (partitions.size() > 1 ? " have " : " has ") + + "been found instead"); } connectionFactory.getLeaders(partitions); return partitions; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 77810e537..83c44e391 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -89,7 +89,7 @@ public class KafkaBinderConfiguration { kafkaMessageChannelBinder.setReplicationFactor(kafkaBinderConfigurationProperties.getReplicationFactor()); kafkaMessageChannelBinder.setRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks()); kafkaMessageChannelBinder.setMaxWait(kafkaBinderConfigurationProperties.getMaxWait()); - + kafkaMessageChannelBinder.setAutoConfigureTopics(kafkaBinderConfigurationProperties.isAutoConfigureTopics()); kafkaMessageChannelBinder.setProducerListener(producerListener); kafkaMessageChannelBinder.setExtendedBindingProperties(kafkaExtendedBindingProperties); return kafkaMessageChannelBinder; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index 19ebdf589..c266595cd 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -45,6 +45,8 @@ public class KafkaBinderConfigurationProperties { private int maxWait = 100; + private boolean autoConfigureTopics = true; + /** * ZK session timeout in milliseconds. */ @@ -204,4 +206,11 @@ public class KafkaBinderConfigurationProperties { this.queueSize = queueSize; } + public boolean isAutoConfigureTopics() { + return autoConfigureTopics; + } + + public void setAutoConfigureTopics(boolean autoConfigureTopics) { + this.autoConfigureTopics = autoConfigureTopics; + } } diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 8c76e9e7e..f8675ebe1 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -16,36 +16,46 @@ package org.springframework.cloud.stream.binder.kafka; +import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.not; import static org.hamcrest.Matchers.nullValue; +import static org.hamcrest.collection.IsArrayContaining.hasItemInArray; +import static org.hamcrest.collection.IsArrayWithSize.arrayWithSize; import static org.hamcrest.collection.IsCollectionWithSize.hasSize; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.util.Arrays; import java.util.Collection; +import java.util.Properties; import java.util.UUID; +import org.hamcrest.collection.IsCollectionWithSize; import org.junit.Before; import org.junit.ClassRule; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.binder.TestUtils; import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.kafka.core.Partition; +import org.springframework.integration.kafka.core.TopicNotFoundException; import org.springframework.integration.kafka.support.ProducerConfiguration; import org.springframework.integration.kafka.support.ProducerMetadata; import org.springframework.integration.kafka.support.ZookeeperConnect; @@ -55,6 +65,11 @@ import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.retry.backoff.FixedBackOffPolicy; +import org.springframework.retry.policy.SimpleRetryPolicy; +import org.springframework.retry.support.RetryTemplate; + +import kafka.admin.AdminUtils; /** @@ -449,6 +464,158 @@ public class KafkaBinderTests extends PartitionCapableBinderTests consumerProperties = createConsumerProperties(); + String testTopicName = "nonexisting" + System.currentTimeMillis(); + try { + binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + fail(); + } + catch (Exception e) { + assertTrue(e instanceof BinderException); + assertThat(e.getCause(), instanceOf(TopicNotFoundException.class)); + assertThat(e.getCause().getMessage(), containsString(testTopicName)); + } + + try { + binder.getConnectionFactory().getPartitions(testTopicName); + fail(); + } + catch (Exception e) { + assertThat(e, instanceOf(TopicNotFoundException.class)); + } + } + + @Test + public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting() throws Exception { + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 5, 1, new Properties()); + + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( + new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), + kafkaTestSupport.getZkConnectString()); + GenericApplicationContext context = new GenericApplicationContext(); + binder.setAutoConfigureTopics(false); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + } + + @Test + public void testAutoConfigureTopicsDisabledFailsIfTopicUnderpartitioned() throws Exception { + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties()); + + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( + new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), + kafkaTestSupport.getZkConnectString()); + GenericApplicationContext context = new GenericApplicationContext(); + binder.setAutoConfigureTopics(false); + + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + try { + binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + } + catch (Exception e) { + assertThat(e, instanceOf(BinderException.class)); + assertThat(e.getCause(), instanceOf(IllegalStateException.class)); + assertThat(e.getCause().getMessage(), + containsString("The number of expected partitions was: 3, but 1 has been found instead")); + } + } + + @Test + public void testAutoConfigureTopicsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception { + + String testTopicName = "existing" + System.currentTimeMillis(); + AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties()); + + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( + new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), + kafkaTestSupport.getZkConnectString()); + GenericApplicationContext context = new GenericApplicationContext(); + binder.setAutoConfigureTopics(false); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + // this consumer must consume from partition 2 + consumerProperties.setInstanceCount(3); + consumerProperties.setInstanceIndex(2); + + Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + + Partition[] listenedPartitions + = TestUtils.getPropertyValue(binding, "endpoint.val$messageListenerContainer.partitions", Partition[].class); + + assertThat(listenedPartitions, arrayWithSize(2)); + assertThat(listenedPartitions, hasItemInArray(new Partition(testTopicName, 2))); + assertThat(listenedPartitions, hasItemInArray(new Partition(testTopicName, 5))); + Collection partitions = binder.getConnectionFactory().getPartitions(testTopicName); + assertThat(partitions, IsCollectionWithSize.hasSize(6)); + binding.unbind(); + } + + @Test + public void testAutoConfigureTopicsEnabledSucceeds() throws Exception { + KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder( + new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(), + kafkaTestSupport.getZkConnectString()); + GenericApplicationContext context = new GenericApplicationContext(); + binder.setAutoConfigureTopics(true); + context.refresh(); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate(); + metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy()); + FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy(); + backOffPolicy.setBackOffPeriod(1000); + metatadataRetrievalRetryOperations.setBackOffPolicy(backOffPolicy); + binder.setMetadataRetryOperations(metatadataRetrievalRetryOperations); + DirectChannel output = new DirectChannel(); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + String testTopicName = "nonexisting" + System.currentTimeMillis(); + Binding binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties); + binding.unbind(); + } + private static class FailingInvocationCountingMessageHandler implements MessageHandler { private int invocationCount = 0; diff --git a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 69051b95e..6a757d6a8 100644 --- a/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -110,7 +110,7 @@ You can use the extensible API to write your own Binder. Spring Cloud Stream uses Spring Boot for configuration, and the Binder abstraction makes it possible for a Spring Cloud Stream application to be flexible in how it connects to middleware. For example, deployers can dynamically choose, at runtime, the destinations (e.g., the Kafka topics or RabbitMQ exchanges) to which channels connect. Such configuration can be provided through external configuration properties and in any form supported by Spring Boot (including application arguments, environment variables, and `application.yml` or `application.properties` files). -In the sink example from the <<_introducing_spring_cloud_stream>> section, setting the application property `spring.cloud.stream.bindings.input.destination` to `raw-sensor-data` will cause it to read from the `raw-sensor-data` Kafka topic, or from a queue bound to the `raw-sensor-data` RabbitMQ exchange. +In the sink example from the <<_introducing_spring_cloud_stream>> section, setting the application property `spring.cloud.stream.bindings.input.destination` to `raw-sensor-data` will cause it to read from the `raw-sensor-data` Kafka topic, or from a queue bound to the `raw-sensor-data` RabbitMQ exchange. Spring Cloud Stream automatically detects and uses a binder found on the classpath. You can easily use different types of middleware with the same code: just include a different binder at build time. @@ -654,7 +654,7 @@ image::kafka-binder.png[width=300,scaledwidth="50%"] The Kafka Binder implementation maps the destination to a Kafka topic. The consumer group maps directly to the same Kafka concept. Spring Cloud Stream does not use the high-level consumer, but implements a similar concept for the simple consumer. - + ==== RabbitMQ Binder @@ -716,12 +716,12 @@ group:: Applies only to inbound bindings. See <>. + -Default: null (indicating an anonymous consumer). +Default: null (indicating an anonymous consumer). contentType:: The content type of the channel. //See <>. + -Default: null (so that no type coercion is performed). +Default: null (so that no type coercion is performed). binder:: The binder used by this binding. See <> for details. @@ -776,26 +776,26 @@ If set, or if `partitionKeyExtractorClass` is set, outbound data on this channel The two options are mutually exclusive. See <>. + -Default: null. +Default: null. partitionKeyExtractorClass:: A `PartitionKeyExtractorStrategy` implementation. If set, or if `partitionKeyExpression` is set, outbound data on this channel will be partitioned, and `partitionCount` must be set to a value greater than 1 to be effective. The two options are mutually exclusive. See <>. + -Default: null. +Default: null. partitionSelectorClass:: A `PartitionSelectorStrategy` implementation. Mutually exclusive with `partitionSelectorExpression`. If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`. + -Default: null. +Default: null. partitionSelectorExpression:: A SpEL expression for customizing partition selection. Mutually exclusive with `partitionSelectorClass`. If neither is set, the partition will be selected as the `hashCode(key) % partitionCount`, where `key` is computed via either `partitionKeyExpression` or `partitionKeyExtractorClass`. + -Default: null. +Default: null. partitionCount:: The number of target partitions for the data, if partitioning is enabled. Must be @@ -803,7 +803,7 @@ Must be On Kafka, interpreted as a hint; the larger of this and the partition count of the target topic is used instead. + -Default: `1`. +Default: `1`. requiredGroups:: A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (e.g., by pre-creating durable queues in RabbitMQ). headerMode:: @@ -976,13 +976,26 @@ spring.cloud.stream.kafka.binder.offsetUpdateCount:: Ignored if `0`. Mutually exclusive with `offsetUpdateTimeWindow`. + -Default: `0`. +Default: `0`. spring.cloud.stream.kafka.binder.requiredAcks:: The number of required acks on the broker. ++ +Default: `1`. spring.cloud.stream.kafka.binder.minPartitionCount:: The minimum number of partitions expected by the consumer if it creates the consumed topic automatically. + Default: `1`. +spring.cloud.stream.kafka.binder.replicationFactor:: + The replication factor of auto-created topics if `autoConfigureTopics` is active. ++ +Default: `1`. +spring.cloud.stream.kafka.binder.autoConfigureTopics:: + If set to `true`, the binder will create new topics or add new partitions to existing topics automatically to match the requirements of the producers and consumers. +If set to `false`, the binder will rely on the topics being already configured. +In the latter case, if the topics do not exist or the partition count is smaller than the expected partition count, the binder will fail to start. +Of note, this setting is independent of the `auto.topic.create.enable` setting of the broker and it does not influence it: if the server is set to auto-create topics, they may be created as part of the metadata retrieval request, with default broker settings. ++ +Default: `true`. ==== Kafka Consumer Properties @@ -1281,7 +1294,7 @@ To get started with creating Spring Cloud Stream applications, visit the https:/ Select Spring Boot version 1.3.4 SNAPSHOT and search or tick the checkbox for Stream Kafka (we will be using Kafka for messaging). Next, create a new class, `GreetingSource`, in the same package as the `GreetingSourceApplication` class. -Give it the following code: +Give it the following code: [source,java] ---- diff --git a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java index 77539d063..876861498 100644 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java +++ b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java @@ -19,6 +19,13 @@ package org.springframework.cloud.stream.test.junit.kafka; import java.util.Properties; +import kafka.server.KafkaConfig; +import kafka.server.KafkaServer; +import kafka.utils.SystemTime$; +import kafka.utils.TestUtils; +import kafka.utils.Utils; +import kafka.utils.ZKStringSerializer$; +import kafka.utils.ZkUtils; import org.I0Itec.zkclient.ZkClient; import org.I0Itec.zkclient.exception.ZkInterruptedException; import org.apache.commons.logging.Log; @@ -28,14 +35,6 @@ import org.junit.Rule; import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport; import org.springframework.util.SocketUtils; -import kafka.server.KafkaConfig; -import kafka.server.KafkaServer; -import kafka.utils.SystemTime$; -import kafka.utils.TestUtils; -import kafka.utils.Utils; -import kafka.utils.ZKStringSerializer$; -import kafka.utils.ZkUtils; - /** * JUnit {@link Rule} that starts an embedded Kafka server (with an associated Zookeeper) @@ -131,6 +130,8 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport