Simplify the logic in UnexpectedPartitionCountHandling classes

Remove the inner interface and classes for UnexpectedPartitionCountHandling
Directly use auto rebalancing flag to control consumer idling

Resolves #135

Adding tests for auto-rebalancing enabled/disabled and under partitioned scenario

correct the logging message for idle consumers

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java

@@ -105,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(), producerHandling());
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, properties.getPartitionCount(), false);
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -129,11 +129,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 			throw new IllegalArgumentException("Instance count cannot be zero");
 		}
 		int partitionCount = properties.getInstanceCount() * properties.getConcurrency();
-
-		UnexpectedPartitionCountHandling unexpectedPartitionCountHandling = properties.getExtension().isAutoRebalanceEnabled() ?
-				consumerIdlingAllowed() : consumerIdlingForbidden();
-
-		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount, unexpectedPartitionCountHandling);
+		createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(name, partitionCount, properties.getExtension().isAutoRebalanceEnabled());
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
 			final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 					this.configurationProperties.getZkSessionTimeout(),
@@ -143,7 +139,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, unexpectedPartitionCountHandling);
+				createTopicAndPartitions(dlqTopic, partitions, properties.getExtension().isAutoRebalanceEnabled());
 				return new KafkaConsumerDestination(name, partitions, dlqTopic);
 			}
 			return new KafkaConsumerDestination(name, partitions);
@@ -152,10 +148,9 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	private void createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(final String topicName, final int partitionCount,
-																	UnexpectedPartitionCountHandling unexpectedPartitionCountHandling) {
-
+																	boolean tolerateLowerPartitionsOnBroker) {
 		if (this.configurationProperties.isAutoCreateTopics() && adminUtilsOperation != null) {
-			createTopicAndPartitions(topicName, partitionCount, unexpectedPartitionCountHandling);
+			createTopicAndPartitions(topicName, partitionCount, tolerateLowerPartitionsOnBroker);
 		}
 		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. " +
@@ -171,7 +166,7 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	 * desired number.
 	 */
 	private void createTopicAndPartitions(final String topicName, final int partitionCount,
-										UnexpectedPartitionCountHandling unexpectPartitonCountHandling) {
+										boolean tolerateLowerPartitionsOnBroker) {
 
 		final ZkUtils zkUtils = ZkUtils.apply(this.configurationProperties.getZkConnectionString(),
 				this.configurationProperties.getZkSessionTimeout(),
@@ -190,8 +185,16 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					if (this.configurationProperties.isAutoAddPartitions()) {
 						adminUtilsOperation.invokeAddPartitions(zkUtils, topicName, effectivePartitionCount, null, false);
 					}
+					else if (tolerateLowerPartitionsOnBroker) {
+						logger.warn("The number of expected partitions was: " + partitionCount + ", but "
+								+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+								+ "There will be " + (effectivePartitionCount - partitionSize) + " consumers");
+					}
 					else {
-						unexpectPartitonCountHandling.handlePartitionCountTooLow(topicName, partitionSize, effectivePartitionCount);
+						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`");
 					}
 				}
 			}
@@ -236,9 +239,8 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 	}
 
 	public Collection<PartitionInfo> getPartitionsForTopic(final int partitionCount,
-														final UnexpectedPartitionCountHandling unexpectedPartitionCountHandling,
+														final boolean tolerateLowerPartitionsOnBroker,
 														final Callable<Collection<PartitionInfo>> callable) {
-
 		try {
 			return this.metadataRetryOperations
 					.execute(new RetryCallback<Collection<PartitionInfo>, Exception>() {
@@ -247,9 +249,18 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 						public Collection<PartitionInfo> doWithRetry(RetryContext context) throws Exception {
 							Collection<PartitionInfo> partitions = callable.call();
 							// do a sanity check on the partition set
+							int partitionSize = partitions.size();
 							if (partitions.size() < partitionCount) {
-								String topic = partitions.isEmpty() ? "unknown" : partitions.iterator().next().topic();
-								unexpectedPartitionCountHandling.handlePartitionCountTooLow(topic, partitions.size(), partitionCount);
+								if (tolerateLowerPartitionsOnBroker) {
+									logger.warn("The number of expected partitions was: " + partitionCount + ", but "
+											+ partitionSize + (partitionSize > 1 ? " have " : " has ") + "been found instead."
+											+ "There will be " + (partitionCount - partitionSize) + "idle consumers");
+								}
+								else {
+									throw new IllegalStateException("The number of expected partitions was: "
+											+ partitionCount + ", but " + partitions.size()
+											+ (partitions.size() > 1 ? " have " : " has ") + "been found instead");
+								}
 							}
 							return partitions;
 						}
@@ -330,50 +341,5 @@ public class KafkaTopicProvisioner implements ProvisioningProvider<ExtendedConsu
 					", dlqName='" + dlqName + '\'' +
 					'}';
 		}
-
 	}
-
-	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);
-	}
-
 }

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

@@ -44,7 +44,6 @@ 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;
@@ -148,7 +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(),
+				false,
 				new Callable<Collection<PartitionInfo>>() {
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {
@@ -213,11 +212,8 @@ 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,
+				extendedConsumerProperties.getExtension().isAutoRebalanceEnabled(),
 				new Callable<Collection<PartitionInfo>>() {
 					@Override
 					public Collection<PartitionInfo> call() throws Exception {

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

@@ -35,7 +35,9 @@ import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.assertj.core.api.Condition;
 import org.junit.Ignore;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.cloud.stream.binder.Binder;
@@ -85,7 +87,10 @@ import static org.junit.Assert.assertTrue;
  * @author Henryk Konsek
  */
 public abstract class KafkaBinderTests extends PartitionCapableBinderTests<AbstractKafkaTestBinder, ExtendedConsumerProperties<KafkaConsumerProperties>,
-						ExtendedProducerProperties<KafkaProducerProperties>> {
+		ExtendedProducerProperties<KafkaProducerProperties>> {
+
+	@Rule
+	public ExpectedException expectedProvisioningException = ExpectedException.none();
 
 	@Override
 	protected ExtendedConsumerProperties<KafkaConsumerProperties> createConsumerProperties() {
@@ -115,10 +120,10 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 	protected abstract ZkUtils getZkUtils(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties);
 
 	protected abstract void invokeCreateTopic(ZkUtils zkUtils, String topic, int partitions,
-												int replicationFactor, Properties topicConfig);
+											int replicationFactor, Properties topicConfig);
 
 	protected abstract int invokePartitionSize(String topic,
-												ZkUtils zkUtils);
+											ZkUtils zkUtils);
 
 	@Test
 	@SuppressWarnings("unchecked")
@@ -1170,7 +1175,7 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoAddPartitionsDisabledFailsIfTopicUnderpartitioned() throws Exception {
+	public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled() throws Exception {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
 
 		final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
@@ -1185,26 +1190,46 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		Binder binder = getBinder(configurationProperties);
 		GenericApplicationContext context = new GenericApplicationContext();
 		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 = null;
-		try {
-			binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(ProvisioningException.class);
-			assertThat(e)
-					.hasMessageContaining("The number of expected partitions was: 3, but 1 has been found instead");
-		}
-		finally {
-			if (binding != null) {
-				binding.unbind();
-			}
+		Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
+		binding.unbind();
+		assertThat(invokePartitionSize(testTopicName, zkUtils)).isEqualTo(1);
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled() throws Exception {
+		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
+
+		final ZkClient zkClient = new ZkClient(configurationProperties.getZkConnectionString(),
+				configurationProperties.getZkSessionTimeout(), configurationProperties.getZkConnectionTimeout(),
+				ZKStringSerializer$.MODULE$);
+
+		final ZkUtils zkUtils = new ZkUtils(zkClient, null, false);
+
+		String testTopicName = "existing" + System.currentTimeMillis();
+		invokeCreateTopic(zkUtils, testTopicName, 1, 1, new Properties());
+		configurationProperties.setAutoAddPartitions(false);
+		Binder binder = getBinder(configurationProperties);
+		GenericApplicationContext context = new GenericApplicationContext();
+		context.refresh();
+
+		DirectChannel output = new DirectChannel();
+		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
+		// this consumer must consume from partition 2
+		consumerProperties.setInstanceCount(3);
+		consumerProperties.setInstanceIndex(2);
+		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
+		expectedProvisioningException.expect(ProvisioningException.class);
+		expectedProvisioningException.expectMessage("The number of expected partitions was: 3, but 1 has been found instead");
+		Binding binding = binder.bindConsumer(testTopicName, "test", output, consumerProperties);
+		if (binding != null) {
+			binding.unbind();
 		}
 	}
 
@@ -1341,11 +1366,11 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 	}
 
 	private KafkaConsumer getKafkaConsumer(Binding binding) {
-		DirectFieldAccessor bindingAccessor = new DirectFieldAccessor((DefaultBinding)binding);
+		DirectFieldAccessor bindingAccessor = new DirectFieldAccessor((DefaultBinding) binding);
 		KafkaMessageDrivenChannelAdapter adapter = (KafkaMessageDrivenChannelAdapter) bindingAccessor.getPropertyValue("lifecycle");
 		DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter);
 		ConcurrentMessageListenerContainer messageListenerContainer = (ConcurrentMessageListenerContainer) adapterAccessor.getPropertyValue("messageListenerContainer");
-		DirectFieldAccessor containerAccessor = new DirectFieldAccessor((ConcurrentMessageListenerContainer)messageListenerContainer);
+		DirectFieldAccessor containerAccessor = new DirectFieldAccessor((ConcurrentMessageListenerContainer) messageListenerContainer);
 		DefaultKafkaConsumerFactory consumerFactory = (DefaultKafkaConsumerFactory) containerAccessor.getPropertyValue("consumerFactory");
 		return (KafkaConsumer) consumerFactory.createConsumer();
 	}
@@ -1472,9 +1497,9 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		input2.setBeanName("test.input2J");
 		Binding<MessageChannel> input2Binding = binder.bindConsumer("partJ.raw.0", "test", input2, consumerProperties);
 
-		output.send(new GenericMessage<>(new byte[] {(byte) 0}));
-		output.send(new GenericMessage<>(new byte[] {(byte) 1}));
-		output.send(new GenericMessage<>(new byte[] {(byte) 2}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 0}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 1}));
+		output.send(new GenericMessage<>(new byte[]{(byte) 2}));
 
 		Message<?> receive0 = receive(input0);
 		assertThat(receive0).isNotNull();
@@ -1533,13 +1558,13 @@ public abstract class KafkaBinderTests extends PartitionCapableBinderTests<Abstr
 		input2.setBeanName("test.input2S");
 		Binding<MessageChannel> input2Binding = binder.bindConsumer("part.raw.0", "test", input2, consumerProperties);
 
-		Message<byte[]> message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[] {2})
+		Message<byte[]> message2 = org.springframework.integration.support.MessageBuilder.withPayload(new byte[]{2})
 				.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "kafkaBinderTestCommonsDelegate")
 				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42)
 				.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build();
 		output.send(message2);
-		output.send(new GenericMessage<>(new byte[] {1}));
-		output.send(new GenericMessage<>(new byte[] {0}));
+		output.send(new GenericMessage<>(new byte[]{1}));
+		output.send(new GenericMessage<>(new byte[]{0}));
 		Message<?> receive0 = receive(input0);
 		assertThat(receive0).isNotNull();
 		Message<?> receive1 = receive(input1);