KGH-309: Allow override of partition count

See: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/309

Provide binders with access to the output channel so it can override the
partition count on the `PartitioningInterceptor` if it detects that the
actual partition count is different to that configured in the producer properties.

Resolves #1421

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

@@ -154,7 +154,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			SubscribableChannel errorChannel = producerProperties.isErrorChannelEnabled()
 					? registerErrorInfrastructure(producerDestination) : null;
 			producerMessageHandler = createProducerMessageHandler(producerDestination, producerProperties,
-					errorChannel);
+					outputChannel, errorChannel);
 			if (producerMessageHandler instanceof InitializingBean) {
 				((InitializingBean) producerMessageHandler).afterPropertiesSet();
 			}
@@ -218,7 +218,35 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	}
 
 	/**
-	 * Creates a {@link MessageHandler} with the ability to send data to the target
+	 * Create a {@link MessageHandler} with the ability to send data to the target
+	 * middleware. If the returned instance is also a {@link Lifecycle}, it will be
+	 * stopped automatically by the binder.
+	 * <p>
+	 * In order to be fully compliant, the {@link MessageHandler} of the binder must
+	 * observe the following headers:
+	 * <ul>
+	 * <li>{@link BinderHeaders#PARTITION_HEADER} - indicates the target partition where
+	 * the message must be sent</li>
+	 * </ul>
+	 * <p>
+	 *
+	 * @param destination the name of the target destination.
+	 * @param producerProperties the producer properties.
+	 * @param channel the channel to bind.
+	 * @param errorChannel the error channel (if enabled, otherwise null). If not null,
+	 * the binder must wire this channel into the producer endpoint so that errors
+	 * are forwarded to it.
+	 * @return the message handler for sending data to the target middleware
+	 * @throws Exception
+	 */
+	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
+			P producerProperties, MessageChannel channel, MessageChannel errorChannel)
+			throws Exception {
+		return createProducerMessageHandler(destination, producerProperties, errorChannel);
+	}
+
+	/**
+	 * Create a {@link MessageHandler} with the ability to send data to the target
 	 * middleware. If the returned instance is also a {@link Lifecycle}, it will be
 	 * stopped automatically by the binder.
 	 * <p>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -42,6 +42,8 @@ public class PartitionHandler {
 
 	private final PartitionSelectorStrategy partitionSelectorStrategy;
 
+	private volatile int partitionCount;
+
 	/**
 	 * Construct a {@code PartitionHandler}.
 	 *
@@ -58,8 +60,18 @@ public class PartitionHandler {
 		this.producerProperties = properties;
 		this.partitionKeyExtractorStrategy = partitionKeyExtractorStrategy;
 		this.partitionSelectorStrategy = partitionSelectorStrategy;
+		this.partitionCount = producerProperties.getPartitionCount();
 	}
 
+	/**
+	 * Set the actual partition count (if different to the configured count).
+	 * @param partitionCount the count.
+	 */
+	public void setPartitionCount(int partitionCount) {
+		this.partitionCount = partitionCount;
+	}
+
+
 	/**
 	 * Determine the partition to which to send this message.
 	 * <p>
@@ -86,10 +98,10 @@ public class PartitionHandler {
 					this.evaluationContext, key, Integer.class);
 		}
 		else {
-			partition = this.partitionSelectorStrategy.selectPartition(key, producerProperties.getPartitionCount());
+			partition = this.partitionSelectorStrategy.selectPartition(key, this.partitionCount);
 		}
 		// protection in case a user selector returns a negative.
-		return Math.abs(partition % producerProperties.getPartitionCount());
+		return Math.abs(partition % this.partitionCount);
 	}
 
 	private Object extractKey(Message<?> message) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -50,7 +50,7 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
-import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -339,7 +339,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 	/**
 	 *
 	 */
-	private abstract class AbstractContentTypeInterceptor extends ChannelInterceptorAdapter {
+	private abstract class AbstractContentTypeInterceptor implements ChannelInterceptor {
 		final MimeType mimeType;
 
 		private AbstractContentTypeInterceptor(String contentType) {
@@ -354,10 +354,7 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 		protected abstract Message<?> doPreSend(Message<?> message, MessageChannel channel);
 	}
 
-	/**
-	 *
-	 */
-	public final class PartitioningInterceptor extends ChannelInterceptorAdapter {
+	public final class PartitioningInterceptor implements ChannelInterceptor {
 
 		private final BindingProperties bindingProperties;
 
@@ -374,6 +371,10 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 					partitionSelectorStrategy);
 		}
 
+		public void setPartitionCount(int partitionCount) {
+			this.partitionHandler.setPartitionCount(partitionCount);
+		}
+
 		@Override
 		public Message<?> preSend(Message<?> message, MessageChannel channel) {
 			if (!message.getHeaders().containsKey(BinderHeaders.PARTITION_OVERRIDE)) {
@@ -392,4 +393,5 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 			}
 		}
 	}
+
 }