GH-509: Rework Topic Autoconfiguration

Fixes #509

- prevent autoconfiguration from reducing the existing partition count and/or replication factor

Split `autoConfigureTopics` in two separate settings, one for `autoCreateTopics`
and one for `autoAddPartitions` (to existing topics)

Renamed `defaultMinPartitionCount` to just `minPartitionCount` on the Kafka binder

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

@@ -34,6 +34,8 @@ import java.util.concurrent.atomic.AtomicInteger;
 
 import kafka.admin.AdminUtils;
 import kafka.api.OffsetRequest;
+import kafka.api.TopicMetadata;
+import kafka.common.ErrorMapping;
 import kafka.serializer.Decoder;
 import kafka.serializer.DefaultDecoder;
 import kafka.utils.ZKStringSerializer$;
@@ -126,7 +128,9 @@ public class KafkaMessageChannelBinder
 
 	public static final DaemonThreadFactory DAEMON_THREAD_FACTORY = new DaemonThreadFactory();
 
-	private boolean autoConfigureTopics = true;
+	private boolean autoCreateTopics = true;
+
+	private boolean autoAddPartitions = false;
 
 	private RetryOperations metadataRetryOperations;
 
@@ -152,7 +156,7 @@ public class KafkaMessageChannelBinder
 
 	private int fetchSize = 1024 * 1024;
 
-	private int defaultMinPartitionCount = 1;
+	private int minPartitionCount = 1;
 
 	private ConnectionFactory connectionFactory;
 
@@ -298,8 +302,8 @@ public class KafkaMessageChannelBinder
 		this.fetchSize = fetchSize;
 	}
 
-	public void setDefaultMinPartitionCount(int defaultMinPartitionCount) {
-		this.defaultMinPartitionCount = defaultMinPartitionCount;
+	public void setMinPartitionCount(int minPartitionCount) {
+		this.minPartitionCount = minPartitionCount;
 	}
 
 	public void setMaxWait(int maxWait) {
@@ -322,12 +326,20 @@ public class KafkaMessageChannelBinder
 		this.zkConnectionTimeout = zkConnectionTimeout;
 	}
 
-	public boolean isAutoConfigureTopics() {
-		return autoConfigureTopics;
+	public boolean isAutoCreateTopics() {
+		return autoCreateTopics;
 	}
 
-	public void setAutoConfigureTopics(boolean autoConfigureTopics) {
-		this.autoConfigureTopics = autoConfigureTopics;
+	public void setAutoCreateTopics(boolean autoCreateTopics) {
+		this.autoCreateTopics = autoCreateTopics;
+	}
+
+	public boolean isAutoAddPartitions() {
+		return autoAddPartitions;
+	}
+
+	public void setAutoAddPartitions(boolean autoAddPartitions) {
+		this.autoAddPartitions = autoAddPartitions;
 	}
 
 	@Override
@@ -381,8 +393,15 @@ public class KafkaMessageChannelBinder
 
 		validateTopicName(name);
 
-		int numPartitions = Math.max(defaultMinPartitionCount, properties.getPartitionCount());
-		Collection<Partition> partitions = ensureTopicCreated(name, numPartitions, replicationFactor);
+		Collection<Partition> partitions = ensureTopicCreated(name, properties.getPartitionCount());
+
+		if (properties.getPartitionCount() < partitions.size()) {
+			if (logger.isInfoEnabled()) {
+				logger.info("The `partitionCount` of the producer for topic " + name + " is " +
+						properties.getPartitionCount() + ", smaller than the actual partition count of " +
+						partitions.size() + " of the topic. The larger number will be used instead.");
+			}
+		}
 
 		topicsInUse.put(name, partitions);
 
@@ -429,28 +448,53 @@ 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 minExpectedPartitions,
-			int replicationFactor) {
+	private Collection<Partition> ensureTopicCreated(final String topicName, final int partitionCount) {
 
 		final ZkClient zkClient = new ZkClient(zkAddress, getZkSessionTimeout(), getZkConnectionTimeout(),
 				ZKStringSerializer$.MODULE$);
 		try {
-			// The following is basically copy/paste from AdminUtils.createTopic() with
-			// createOrUpdateTopicPartitionAssignmentPathInZK(..., update=true)
 			final Properties topicConfig = new Properties();
-			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;
+			TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(topicName, zkClient);
+			if (topicMetadata.errorCode() == ErrorMapping.NoError()) {
+				// only consider minPartitionCount for resizing if autoAddPartitions is true
+				int effectivePartitionCount = isAutoAddPartitions() ? Math.max(minPartitionCount, partitionCount)
+						: partitionCount;
+				if (topicMetadata.partitionsMetadata().size() < effectivePartitionCount) {
+					if (isAutoAddPartitions()) {
+						AdminUtils.addPartitions(zkClient, topicName, effectivePartitionCount, null, false,
+								new Properties());
 					}
-				});
+					else {
+						int topicSize = topicMetadata.partitionsMetadata().size();
+						throw new BinderException("The number of expected partitions was: " + partitionCount
+								+ ", but " + topicSize + (topicSize > 1 ? " have " : " has ") + "been found instead." +
+								"Consider either increasing the partition count of the topic or enabling `autoAddPartitions`");
+					}
+				}
+			}
+			else if (topicMetadata.errorCode() == ErrorMapping.UnknownTopicOrPartitionCode()) {
+				if (isAutoCreateTopics()) {
+					Seq<Object> brokerList = ZkUtils.getSortedBrokerList(zkClient);
+					// always consider minPartitionCount for topic creation
+					int effectivePartitionCount = Math.max(this.minPartitionCount, partitionCount);
+					final scala.collection.Map<Object, Seq<Object>> replicaAssignment = AdminUtils
+							.assignReplicasToBrokers(brokerList, effectivePartitionCount, 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;
+						}
+					});
+				}
+				else {
+					throw new BinderException("Topic " + topicName + " does not exist");
+				}
+			}
+			else {
+				throw new BinderException("Error fetching Kafka topic metadata: ",
+						ErrorMapping.exceptionFor(topicMetadata.errorCode()));
 			}
 			try {
 				Collection<Partition> partitions = metadataRetryOperations
@@ -460,11 +504,11 @@ public class KafkaMessageChannelBinder
 							public Collection<Partition> doWithRetry(RetryContext context) throws Exception {
 								connectionFactory.refreshMetadata(Collections.singleton(topicName));
 								Collection<Partition> partitions = connectionFactory.getPartitions(topicName);
-								if (partitions.size() < minExpectedPartitions) {
-									throw new IllegalStateException(
-											"The number of expected partitions was: " + minExpectedPartitions + ", but "
-													+ partitions.size() + (partitions.size() > 1 ? " have " : " has ")
-													+ "been found instead");
+								// do a sanity check on the partition set
+								if (partitions.size() < partitionCount) {
+									throw new IllegalStateException("The number of expected partitions was: "
+											+ partitionCount + ", but " + partitions.size()
+											+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
 								}
 								connectionFactory.getLeaders(partitions);
 								return partitions;
@@ -487,26 +531,25 @@ public class KafkaMessageChannelBinder
 			ExtendedConsumerProperties<KafkaConsumerProperties> properties, String group, long referencePoint) {
 
 		validateTopicName(name);
-		int instance = properties.getInstanceCount();
-		if (instance == 0) {
+		if (properties.getInstanceCount() == 0) {
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
-		int numPartitions = Math.max(this.defaultMinPartitionCount, instance * properties.getConcurrency());
-		Collection<Partition> allPartitions = ensureTopicCreated(name, numPartitions, replicationFactor);
+		Collection<Partition> allPartitions = ensureTopicCreated(name,
+				properties.getInstanceCount() * properties.getConcurrency());
 
 		Decoder<byte[]> valueDecoder = new DefaultDecoder(null);
 		Decoder<byte[]> keyDecoder = new DefaultDecoder(null);
 
 		Collection<Partition> listenedPartitions;
 
-		if (instance == 1) {
+		if (properties.getInstanceCount() == 1) {
 			listenedPartitions = allPartitions;
 		}
 		else {
 			listenedPartitions = new ArrayList<>();
 			for (Partition partition : allPartitions) {
 				// divide partitions across modules
-				if ((partition.getId() % instance) == properties.getInstanceIndex()) {
+				if ((partition.getId() % properties.getInstanceCount()) == properties.getInstanceIndex()) {
 					listenedPartitions.add(partition);
 				}
 			}

spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java

@@ -84,12 +84,13 @@ public class KafkaBinderConfiguration {
 		kafkaMessageChannelBinder.setZkConnectionTimeout(kafkaBinderConfigurationProperties.getZkConnectionTimeout());
 
 		kafkaMessageChannelBinder.setFetchSize(kafkaBinderConfigurationProperties.getFetchSize());
-		kafkaMessageChannelBinder.setDefaultMinPartitionCount(kafkaBinderConfigurationProperties.getMinPartitionCount());
+		kafkaMessageChannelBinder.setMinPartitionCount(kafkaBinderConfigurationProperties.getMinPartitionCount());
 		kafkaMessageChannelBinder.setQueueSize(kafkaBinderConfigurationProperties.getQueueSize());
 		kafkaMessageChannelBinder.setReplicationFactor(kafkaBinderConfigurationProperties.getReplicationFactor());
 		kafkaMessageChannelBinder.setRequiredAcks(kafkaBinderConfigurationProperties.getRequiredAcks());
 		kafkaMessageChannelBinder.setMaxWait(kafkaBinderConfigurationProperties.getMaxWait());
-		kafkaMessageChannelBinder.setAutoConfigureTopics(kafkaBinderConfigurationProperties.isAutoConfigureTopics());
+		kafkaMessageChannelBinder.setAutoCreateTopics(kafkaBinderConfigurationProperties.isAutoCreateTopics());
+		kafkaMessageChannelBinder.setAutoAddPartitions(kafkaBinderConfigurationProperties.isAutoAddPartitions());
 		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,7 +45,9 @@ public class KafkaBinderConfigurationProperties {
 
 	private int maxWait = 100;
 
-	private boolean autoConfigureTopics = true;
+	private boolean autoCreateTopics = true;
+
+	private boolean autoAddPartitions = false;
 
 	/**
 	 * ZK session timeout in milliseconds.
@@ -206,11 +208,19 @@ public class KafkaBinderConfigurationProperties {
 		this.queueSize = queueSize;
 	}
 
-	public boolean isAutoConfigureTopics() {
-		return autoConfigureTopics;
+	public boolean isAutoCreateTopics() {
+		return autoCreateTopics;
 	}
 
-	public void setAutoConfigureTopics(boolean autoConfigureTopics) {
-		this.autoConfigureTopics = autoConfigureTopics;
+	public void setAutoCreateTopics(boolean autoCreateTopics) {
+		this.autoCreateTopics = autoCreateTopics;
+	}
+
+	public boolean isAutoAddPartitions() {
+		return autoAddPartitions;
+	}
+
+	public void setAutoAddPartitions(boolean autoAddPartitions) {
+		this.autoAddPartitions = autoAddPartitions;
 	}
 }

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

@@ -36,6 +36,8 @@ import java.util.Collection;
 import java.util.Properties;
 import java.util.UUID;
 
+import kafka.admin.AdminUtils;
+import kafka.api.TopicMetadata;
 import org.hamcrest.collection.IsCollectionWithSize;
 import org.junit.Before;
 import org.junit.ClassRule;
@@ -69,8 +71,6 @@ import org.springframework.retry.backoff.FixedBackOffPolicy;
 import org.springframework.retry.policy.SimpleRetryPolicy;
 import org.springframework.retry.support.RetryTemplate;
 
-import kafka.admin.AdminUtils;
-
 
 /**
  * Integration tests for the {@link KafkaMessageChannelBinder}.
@@ -465,13 +465,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 	}
 
 	@Test
-	public void testAutoConfigureTopicsDisabledFailsIfTopicMissing() throws Exception {
+	public void testAutoCreateTopicsDisabledFailsIfTopicMissing() throws Exception {
 
 		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
 				new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(),
 				kafkaTestSupport.getZkConnectString());
 		GenericApplicationContext context = new GenericApplicationContext();
-		binder.setAutoConfigureTopics(false);
+		binder.setAutoCreateTopics(false);
 		context.refresh();
 		binder.setApplicationContext(context);
 		binder.afterPropertiesSet();
@@ -490,8 +490,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		}
 		catch (Exception e) {
 			assertTrue(e instanceof BinderException);
-			assertThat(e.getCause(), instanceOf(TopicNotFoundException.class));
-			assertThat(e.getCause().getMessage(), containsString(testTopicName));
+			assertThat(e.getMessage(), containsString("Topic " + testTopicName  + " does not exist"));
 		}
 
 		try {
@@ -513,7 +512,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 				new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(),
 				kafkaTestSupport.getZkConnectString());
 		GenericApplicationContext context = new GenericApplicationContext();
-		binder.setAutoConfigureTopics(false);
+		binder.setAutoCreateTopics(false);
 		context.refresh();
 		binder.setApplicationContext(context);
 		binder.afterPropertiesSet();
@@ -524,7 +523,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 	}
 
 	@Test
-	public void testAutoConfigureTopicsDisabledFailsIfTopicUnderpartitioned() throws Exception {
+	public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception {
 
 		String testTopicName = "existing"  + System.currentTimeMillis();
 		AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 1, 1, new Properties());
@@ -533,7 +532,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 				new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(),
 				kafkaTestSupport.getZkConnectString());
 		GenericApplicationContext context = new GenericApplicationContext();
-		binder.setAutoConfigureTopics(false);
+		binder.setAutoAddPartitions(false);
 
 		context.refresh();
 		binder.setApplicationContext(context);
@@ -548,14 +547,13 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 		}
 		catch (Exception e) {
 			assertThat(e, instanceOf(BinderException.class));
-			assertThat(e.getCause(), instanceOf(IllegalStateException.class));
-			assertThat(e.getCause().getMessage(),
+			assertThat(e.getMessage(),
 					containsString("The number of expected partitions was: 3, but 1 has been found instead"));
 		}
 	}
 
 	@Test
-	public void testAutoConfigureTopicsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception {
+	public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly() throws Exception {
 
 		String testTopicName = "existing"  + System.currentTimeMillis();
 		AdminUtils.createTopic(kafkaTestSupport.getZkClient(), testTopicName, 6, 1, new Properties());
@@ -564,7 +562,7 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 				new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(),
 				kafkaTestSupport.getZkConnectString());
 		GenericApplicationContext context = new GenericApplicationContext();
-		binder.setAutoConfigureTopics(false);
+		binder.setAutoAddPartitions(false);
 		RetryTemplate metatadataRetrievalRetryOperations = new RetryTemplate();
 		metatadataRetrievalRetryOperations.setRetryPolicy(new SimpleRetryPolicy());
 		FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
@@ -594,12 +592,12 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 	}
 
 	@Test
-	public void testAutoConfigureTopicsEnabledSucceeds() throws Exception {
+	public void testAutoCreateTopicsEnabledSucceeds() throws Exception {
 		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
 				new ZookeeperConnect(kafkaTestSupport.getZkConnectString()), kafkaTestSupport.getBrokerAddress(),
 				kafkaTestSupport.getZkConnectString());
 		GenericApplicationContext context = new GenericApplicationContext();
-		binder.setAutoConfigureTopics(true);
+		binder.setAutoCreateTopics(true);
 		context.refresh();
 		binder.setApplicationContext(context);
 		binder.afterPropertiesSet();
@@ -633,4 +631,62 @@ public class KafkaBinderTests extends PartitionCapableBinderTests<KafkaTestBinde
 			return invocationCount;
 		}
 	}
+
+	@Test
+	public void testPartitionCountNotReduced() 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.setAutoAddPartitions(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();
+		Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
+		binding.unbind();
+		TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient());
+		assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6));
+	}
+
+	@Test
+	public void testPartitionCountIncreasedIfAutoAddPartitionsSet() 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());
+
+		binder.setMinPartitionCount(6);
+
+		GenericApplicationContext context = new GenericApplicationContext();
+		binder.setAutoAddPartitions(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();
+		Binding<?> binding = binder.doBindConsumer(testTopicName, "test", output, consumerProperties);
+		binding.unbind();
+		TopicMetadata topicMetadata = AdminUtils.fetchTopicMetadataFromZk(testTopicName, kafkaTestSupport.getZkClient());
+		assertThat(topicMetadata.partitionsMetadata().size(), equalTo(6));
+	}
 }

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

@@ -62,7 +62,7 @@ public class KafkaTestBinder extends
 			binder.setApplicationContext(context);
 			binder.setFetchSize(binderConfiguration.getFetchSize());
 			binder.setMaxWait(binderConfiguration.getMaxWait());
-			binder.setDefaultMinPartitionCount(binderConfiguration.getMinPartitionCount());
+			binder.setMinPartitionCount(binderConfiguration.getMinPartitionCount());
 			binder.afterPropertiesSet();
 			this.setBinder(binder);
 		}

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

@@ -26,6 +26,9 @@ import static org.junit.Assert.assertThat;
 
 import java.util.Arrays;
 
+import org.junit.Ignore;
+import org.junit.Test;
+
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -40,9 +43,6 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
 
-import org.junit.Ignore;
-import org.junit.Test;
-
 /**
  * @author Marius Bogoevici
  * @author David Turanski
@@ -59,7 +59,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		properties.setHeaderMode(HeaderMode.raw);
 		properties.setPartitionKeyExtractorClass(RawKafkaPartitionTestSupport.class);
 		properties.setPartitionSelectorClass(RawKafkaPartitionTestSupport.class);
-		properties.setPartitionCount(3);
+		properties.setPartitionCount(6);
 
 		DirectChannel output = new DirectChannel();
 		output.setBeanName("test.output");
@@ -113,7 +113,7 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests {
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload[0]"));
 		properties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()"));
-		properties.setPartitionCount(3);
+		properties.setPartitionCount(6);
 		properties.setHeaderMode(HeaderMode.raw);
 
 		DirectChannel output = new DirectChannel();

spring-cloud-stream-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -982,20 +982,29 @@ spring.cloud.stream.kafka.binder.requiredAcks::
 +
 Default: `1`.
 spring.cloud.stream.kafka.binder.minPartitionCount::
-  The minimum number of partitions expected by the consumer if it creates the consumed topic automatically.
+  Effective only if `autoCreateTopics` or `autoAddPartitions` is set.
+  The global minimum number of partitions that the binder will configure on topics on which it produces/consumes data.
+It can be superseded by the `partitionCount` setting of the producer or by the value of
+`instanceCount` * `concurrency` settings of the producer (if either is larger).
 +
 Default: `1`.
 spring.cloud.stream.kafka.binder.replicationFactor::
-  The replication factor of auto-created topics if `autoConfigureTopics` is active.
+  The replication factor of auto-created topics if `autoCreateTopics` 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.
+spring.cloud.stream.kafka.binder.autoCreateTopics::
+  If set to `true`, the binder will create new topics automatically.
 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.
+In the latter case, if the topics do not exist, 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`.
+spring.cloud.stream.kafka.binder.autoAddPartitions::
+  If set to `true`, the binder will create add new partitions if required.
+If set to `false`, the binder will rely on the partition size of the topic being already configured.
+If the partition count of the target topic is smaller than the expected value, the binder will fail to start.
+Default: `false`.
+
 
 ==== Kafka Consumer Properties
 
@@ -1168,7 +1177,7 @@ When scaling up Spring Cloud Stream applications, each instance can receive info
 Spring Cloud Stream does this through the `spring.cloud.stream.instanceCount` and `spring.cloud.stream.instanceIndex` properties.
 For example, if there are three instances of a HDFS sink application, all three instances will have `spring.cloud.stream.instanceCount` set to `3`, and the individual applications will have `spring.cloud.stream.instanceIndex` set to `0`, `1`, and `2`, respectively.
 
-When Spring Cloud Stream applications are deployed via Spring Cloud Dataflow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly.
+When Spring Cloud Stream applications are deployed via Spring Cloud Data Flow, these properties are configured automatically; when Spring Cloud Stream applications are launched independently, these properties must be set correctly.
 By default, `spring.cloud.stream.instanceCount` is `1`, and `spring.cloud.stream.instanceIndex` is `0`.
 
 In a scaled-up scenario, correct configuration of these two properties is important for addressing partitioning behavior (see below) in general, and the two properties are always required by certain binders (e.g., the Kafka binder) in order to ensure that data are split correctly across multiple consumer instances.
@@ -1202,6 +1211,15 @@ This can be customized on the binding, either by setting a SpEL expression to be
 
 Additional properties can be configured for more advanced scenarios, as described in the following section.
 
+[NOTE]
+====
+The Kafka binder will use the `partitionCount` setting as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used).
+Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value will be used.
+If a topic already exists with a smaller partition count and `autoAddPartitions` is disabled (the default), then the binder will fail to start.
+If a topic already exists with a smaller partition count and `autoAddPartitions` is enabled, new partitions will be added.
+If a topic already exists with a larger number of partitions than the maximum of (`minPartitionCount` and `partitionCount`), the existing partition count will be used.
+====
+
 ===== Configuring Input Bindings for Partitioning
 
 An input binding is configured to receive partitioned data by setting its `partitioned` property, as well as the `instanceIndex` and `instanceCount` properties on the application itself, as in the following example:

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-2016 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.
@@ -22,10 +22,15 @@ package org.springframework.cloud.stream.binder;
  * likely a configuration error).
  *
  * @author Gary Russell
+ * @author Marius Bogoevici
  */
 @SuppressWarnings("serial")
 public class BinderException extends RuntimeException {
 
+	public BinderException(String message) {
+		super(message);
+	}
+
 	public BinderException(String message, Throwable cause) {
 		super(message, cause);
 	}