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`

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<String, Collection<Partition>> 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<Partition> ensureTopicCreated(final String topicName, final int numPartitions,
+	private Collection<Partition> 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<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
-			final scala.collection.Map<Object, Seq<Object>> replicaAssignment = AdminUtils
-					.assignReplicasToBrokers(brokerList, numPartitions, replicationFactor, -1, -1);
-			retryOperations.execute(new RetryCallback<Object, RuntimeException>() {
+			if (isAutoConfigureTopics()) {
+				Seq<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
+				final scala.collection.Map<Object, Seq<Object>> replicaAssignment = AdminUtils
+						.assignReplicasToBrokers(brokerList, minExpectedPartitions, replicationFactor, -1, -1);
+				metadataRetryOperations.execute(new RetryCallback<Object, RuntimeException>() {
 
-				@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<Partition> partitions = retryOperations
+				Collection<Partition> partitions = metadataRetryOperations
 						.execute(new RetryCallback<Collection<Partition>, Exception>() {
 
 							@Override
 							public Collection<Partition> doWithRetry(RetryContext context) throws Exception {
 								connectionFactory.refreshMetadata(Collections.singleton(topicName));
 								Collection<Partition> 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;

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;

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;
+	}
 }

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<KafkaTestBinde
 		producerBinding.unbind();
 	}
 
+	@Test
+	public void testAutoConfigureTopicsDisabledFailsIfTopicMissing() throws Exception {
+
+		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();
+		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<KafkaConsumerProperties> 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<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		Binding<MessageChannel> 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<KafkaConsumerProperties> 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<KafkaConsumerProperties> 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<Partition> 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<KafkaConsumerProperties> 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;

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 <<consumer-groups,Consumer Groups>>.
 +
-Default: null (indicating an anonymous consumer). 
+Default: null (indicating an anonymous consumer).
 contentType::
     The content type of the channel.
 //See <<content type management>>.
 +
-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 <<multiple-binders>> 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 <<partitioning>>.
 +
-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 <<partitioning>>.
 +
-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]
 ----

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<String
 						log.debug("Creating Kafka server");
 						Properties brokerConfigProperties = brokerConfig;
 						brokerConfig.put("zookeeper.connect", zookeeper.getConnectString());
+						brokerConfig.put("auto.create.topics.enable", "false");
+						brokerConfig.put("delete.topic.enable", "true");
 						kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$);
 						log.debug("Created Kafka server at " + kafkaServer.config().hostName() + ":" + kafkaServer.config().port());
 					}