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 d77852e88..15be7dda5 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
@@ -20,6 +20,8 @@ import java.util.Collection;
 import java.util.Properties;
 import java.util.concurrent.Callable;
 
+import kafka.common.ErrorMapping;
+import kafka.utils.ZkUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.common.PartitionInfo;
@@ -47,15 +49,13 @@ import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import kafka.common.ErrorMapping;
-import kafka.utils.ZkUtils;
-
 /**
  * Kafka implementation for {@link ProvisioningProvider}
  *
  * @author Soby Chacko
  * @author Gary Russell
  * @author Ilayaperumal Gopinathan
+ * @author Simon Flandergan
  */
 public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsumerProperties<KafkaConsumerProperties>,
 		ExtendedProducerProperties<KafkaProducerProperties>>, InitializingBean {
@@ -75,7 +75,6 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	/**
-	 *
 	 * @param metadataRetryOperations the retry configuration
 	 */
 	public void setMetadataRetryOperations(RetryOperations metadataRetryOperations) {
@@ -106,7 +105,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			this.logger.info("Using kafka topic for outbound: " + name);
 		}
 		KafkaTopicUtils.validateTopicName(name);
-		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount());
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount(), producerHandling());
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -130,7 +129,11 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
 		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
-		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount);
+
+		UnexpectedPartitionCountHandling unexpectedPartitionCountHandling = properties.getExtension().isAutoRebalanceEnabled() ?
+				consumerIdlingAllowed() : consumerIdlingForbidden();
+
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount, unexpectedPartitionCountHandling);
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -140,7 +143,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			if (properties.getExtension().isEnableDlq() && !anonymous) {
 				String dlqTopic = StringUtils.hasText(properties.getExtension().getDlqName()) ?
 						properties.getExtension().getDlqName() : "error." + name + "." + group;
-				createTopicAndPartitions(dlqTopic, partitions);
+				createTopicAndPartitions(dlqTopic, partitions, unexpectedPartitionCountHandling);
 				return new KafkaConsumerDestination(name, partitions, dlqTopic);
 			}
 			return new KafkaConsumerDestination(name, partitions);
@@ -148,9 +151,11 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		return new KafkaConsumerDestination(name);
 	}
 
-	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount) {
+	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount,
+																	UnexpectedPartitionCountHandling unexpectedPartitionCountHandling) {
+
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
-			createTopicAndPartitions(topicName, partitionCount);
+			createTopicAndPartitions(topicName, partitionCount, unexpectedPartitionCountHandling);
 		}
 		else if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation == null) {
 			this.logger.warn("Auto creation of topics is enabled, but Kafka AdminUtils class is not present on the classpath. " +
@@ -165,7 +170,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	 * Creates a Kafka topic if needed, or try to increase its partition count to the
 	 * desired number.
 	 */
-	private void createTopicAndPartitions(final String topicName, final int partitionCount) {
+	private void createTopicAndPartitions(final String topicName, final int partitionCount,
+										UnexpectedPartitionCountHandling unexpectPartitonCountHandling) {
+
 		final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 				this.configurationProperties.getZkSessionTimeout(),
 				this.configurationProperties.getZkConnectionTimeout(),
@@ -184,10 +191,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 						adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
 					}
 					else {
-						throw new ProvisioningException("The number of expected partitions was: " + partitionCount + ", but "
-								+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
-								+ "Consider either increasing the partition count of the topic or enabling " +
-								"`autoAddPartitions`");
+						unexpectPartitonCountHandling.handlePartitionCountTooLow(topicName, partitionSize, effectivePartitionCount);
 					}
 				}
 			}
@@ -231,7 +235,10 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 		}
 	}
 
-	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount, final Callable<Collection<PartitionInfo>> callable) {
+	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount,
+														final UnexpectedPartitionCountHandling unexpectedPartitionCountHandling,
+														final Callable<Collection<PartitionInfo>> callable) {
+
 		try {
 			return this.metadataRetryOperations
 					.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
@@ -241,9 +248,8 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 							Collection<PartitionInfo> partitions = callable.call();
 							// 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");
+								String topic = partitions.isEmpty() ? "unknown" : partitions.iterator().next().topic();
+								unexpectedPartitionCountHandling.handlePartitionCountTooLow(topic, partitions.size(), partitionCount);
 							}
 							return partitions;
 						}
@@ -327,4 +333,47 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 
 	}
 
+	public UnexpectedPartitionCountHandling consumerIdlingAllowed() {
+		return new UnexpectedPartitionCountHandling() {
+
+			@Override
+			public void handlePartitionCountTooLow(String topicName, int partitionSize, int effectivePartitionCount) {
+				logger.warn(String.format("There is only %s partitions on topic %s, but %s consumers. %s consumer will be idle",
+						partitionSize, topicName, effectivePartitionCount, effectivePartitionCount - partitionSize));
+			}
+		};
+	}
+
+	public UnexpectedPartitionCountHandling consumerIdlingForbidden() {
+		return new UnexpectedPartitionCountHandling() {
+
+			@Override
+			public void handlePartitionCountTooLow(String topicName, int partitionSize, int effectivePartitionCount) {
+				throw new ProvisioningException("The number of expected partitions was: " + partitionSize + ", but "
+						+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+						+ "Consider either increasing the partition count of the topic or enabling " +
+						"`autoAddPartitions`");
+			}
+		};
+	}
+
+	public UnexpectedPartitionCountHandling producerHandling() {
+		return new UnexpectedPartitionCountHandling() {
+
+			@Override
+			public void handlePartitionCountTooLow(String topicName, int partitionSize, int effectivePartitionCount) {
+				throw new ProvisioningException("The number of expected partitions was: " + partitionSize + ", but "
+						+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+						+ "Consider either increasing the partition count of the topic or enabling " +
+						"`autoAddPartitions`");
+			}
+
+		};
+	}
+
+	public interface UnexpectedPartitionCountHandling {
+
+		void handlePartitionCountTooLow(String topicName, int partitionSize, int effectivePartitionCount);
+	}
+
 }
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 d2ec8e4ab..e67a7e0c3 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
@@ -44,6 +44,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
 import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.UnexpectedPartitionCountHandling;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.context.Lifecycle;
@@ -146,6 +147,7 @@ public class KafkaMessageChannelBinder extends
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) throws Exception {
 		final DefaultKafkaProducerFactory<byte[], byte[]> producerFB = getProducerFactory(producerProperties);
 		Collection<PartitionInfo> partitions = provisioningProvider.getPartitionsForTopic(producerProperties.getPartitionCount(),
+				provisioningProvider.producerHandling(),
 				new Callable<Collection<PartitionInfo>>() {
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {
@@ -210,7 +212,11 @@ public class KafkaMessageChannelBinder extends
 		final ConsumerFactory<?, ?> consumerFactory = createKafkaConsumerFactory(anonymous, consumerGroup, extendedConsumerProperties);
 		int partitionCount = extendedConsumerProperties.getInstanceCount() * extendedConsumerProperties.getConcurrency();
 
+		UnexpectedPartitionCountHandling unexpedPartitionCountHandling = extendedConsumerProperties.getExtension().isAutoRebalanceEnabled()? provisioningProvider.consumerIdlingAllowed()
+				: provisioningProvider.consumerIdlingForbidden();
+		
 		Collection<PartitionInfo> allPartitions = provisioningProvider.getPartitionsForTopic(partitionCount,
+				unexpedPartitionCountHandling,
 				new Callable<Collection<PartitionInfo>>() {
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {
@@ -239,8 +245,8 @@ public class KafkaMessageChannelBinder extends
 		final TopicPartitionInitialOffset[] topicPartitionInitialOffsets = getTopicPartitionInitialOffsets(
 				listenedPartitions);
 		final ContainerProperties containerProperties =
-				anonymous || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ? new ContainerProperties(destination.getName())
-						: new ContainerProperties(topicPartitionInitialOffsets);
+				anonymous || extendedConsumerProperties.getExtension().isAutoRebalanceEnabled() ? 
+						new ContainerProperties(destination.getName()) : new ContainerProperties(topicPartitionInitialOffsets);
 		int concurrency = Math.min(extendedConsumerProperties.getConcurrency(), listenedPartitions.size());
 		final ConcurrentMessageListenerContainer<?, ?> messageListenerContainer =
 				new ConcurrentMessageListenerContainer(