Partition handling refactoring

Moved partition calculation from AbstractBinder to
its own class PartitionHandler.

Updated per review feedback

Refactored bean creation
Added synchronization around bean creation

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

@@ -36,6 +36,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.AbstractBinder;
 import org.springframework.cloud.stream.binder.BinderException;
 import org.springframework.cloud.stream.binder.BinderHeaders;
@@ -45,6 +46,7 @@ import org.springframework.cloud.stream.binder.DefaultBinding;
 import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
 import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
 import org.springframework.cloud.stream.binder.MessageValues;
+import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.http.MediaType;
 import org.springframework.integration.channel.FixedSubscriberChannel;
 import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -164,12 +166,6 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 	private static final boolean DEFAULT_SYNC_PRODUCER = false;
 
-	private static final StartOffset DEFAULT_START_OFFSET = StartOffset.latest;
-
-	private RetryOperations retryOperations;
-
-	private Map<String, Collection<Partition>> topicsInUse = new HashMap<>();
-
 	protected static final Set<Object> PRODUCER_COMPRESSION_PROPERTIES = new HashSet<Object>(
 			Arrays.asList(new String[] {
 				KafkaMessageChannelBinder.COMPRESSION_CODEC,
@@ -207,6 +203,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
 			.addAll(PRODUCER_COMPRESSION_PROPERTIES)
 			.build();
 
+	private RetryOperations retryOperations;
+
+	private final Map<String, Collection<Partition>> topicsInUse = new HashMap<>();
+
 	private final EmbeddedHeadersMessageConverter embeddedHeadersMessageConverter = new
 			EmbeddedHeadersMessageConverter();
 
@@ -830,8 +830,6 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 	private class SendingHandler extends AbstractMessageHandler {
 
-		private final PartitioningMetadata partitioningMetadata;
-
 		private final AtomicInteger roundRobinCount = new AtomicInteger();
 
 		private final String topicName;
@@ -840,21 +838,24 @@ public class KafkaMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 		private final ProducerConfiguration<byte[], byte[]> producerConfiguration;
 
+		private final PartitionHandler partitionHandler;
 
 		private SendingHandler(String topicName, KafkaPropertiesAccessor properties, int numberOfPartitions,
 				ProducerConfiguration<byte[], byte[]> producerConfiguration) {
 			this.topicName = topicName;
 			this.numberOfKafkaPartitions = numberOfPartitions;
-			this.partitioningMetadata = new PartitioningMetadata(properties, numberOfPartitions);
-			this.setBeanFactory(KafkaMessageChannelBinder.this.getBeanFactory());
+			ConfigurableListableBeanFactory beanFactory = KafkaMessageChannelBinder.this.getBeanFactory();
+			this.setBeanFactory(beanFactory);
 			this.producerConfiguration = producerConfiguration;
+			this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector,
+					properties, numberOfPartitions);
 		}
 
 		@Override
 		protected void handleMessageInternal(Message<?> message) throws Exception {
 			int targetPartition;
-			if (partitioningMetadata.isPartitionedModule()) {
-				targetPartition = determinePartition(message, partitioningMetadata);
+			if (this.partitionHandler.isPartitionedModule()) {
+				targetPartition = this.partitionHandler.determinePartition(message);
 			}
 			else {
 				targetPartition = roundRobin() % numberOfKafkaPartitions;

spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java

@@ -67,6 +67,7 @@ import org.springframework.cloud.stream.binder.Binding;
 import org.springframework.cloud.stream.binder.DefaultBinding;
 import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
 import org.springframework.cloud.stream.binder.MessageValues;
+import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.context.Lifecycle;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.io.Resource;
@@ -791,13 +792,15 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 		private final String replyTo;
 
-		private final PartitioningMetadata partitioningMetadata;
+		private final PartitionHandler partitionHandler;
 
 		private SendingHandler(MessageHandler delegate, String replyTo, RabbitPropertiesAccessor properties) {
 			this.delegate = delegate;
 			this.replyTo = replyTo;
-			this.partitioningMetadata = new PartitioningMetadata(properties, properties.getNextModuleCount());
-			this.setBeanFactory(RabbitMessageChannelBinder.this.getBeanFactory());
+			ConfigurableListableBeanFactory beanFactory = RabbitMessageChannelBinder.this.getBeanFactory();
+			this.setBeanFactory(beanFactory);
+			this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector,
+					properties, properties.getNextModuleCount());
 		}
 
 		@Override
@@ -807,8 +810,9 @@ public class RabbitMessageChannelBinder extends AbstractBinder<MessageChannel> {
 			if (this.replyTo != null) {
 				messageToSend.put(AmqpHeaders.REPLY_TO, this.replyTo);
 			}
-			if (this.partitioningMetadata.isPartitionedModule()) {
-				messageToSend.put(PARTITION_HEADER, determinePartition(message, this.partitioningMetadata));
+			if (this.partitionHandler.isPartitionedModule()) {
+				messageToSend.put(PARTITION_HEADER,
+						this.partitionHandler.determinePartition(message));
 			}
 
 			this.delegate.handleMessage(messageToSend.toMessage(getMessageBuilderFactory()));

spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java

@@ -26,6 +26,7 @@ import java.util.Set;
 import java.util.UUID;
 
 import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.BinderPropertyKeys;
 import org.springframework.cloud.stream.binder.Binding;
@@ -33,6 +34,7 @@ import org.springframework.cloud.stream.binder.DefaultBinding;
 import org.springframework.cloud.stream.binder.DefaultBindingPropertiesAccessor;
 import org.springframework.cloud.stream.binder.EmbeddedHeadersMessageConverter;
 import org.springframework.cloud.stream.binder.MessageValues;
+import org.springframework.cloud.stream.binder.PartitionHandler;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisOperations;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -301,17 +303,19 @@ public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel> {
 
 		private final String bindingName;
 
-		private final PartitioningMetadata partitioningMetadata;
-
 		private final RedisPropertiesAccessor accessor;
 
 		private final Map<String, RedisQueueOutboundChannelAdapter> adapters = new HashMap<>();
 
+		private final PartitionHandler partitionHandler;
+
 		private SendingHandler(String bindingName, RedisPropertiesAccessor properties) {
 			this.bindingName = bindingName;
 			this.accessor = properties;
-			this.partitioningMetadata = new PartitioningMetadata(properties, properties.getNextModuleCount());
-			this.setBeanFactory(RedisMessageChannelBinder.this.getBeanFactory());
+			ConfigurableListableBeanFactory beanFactory = RedisMessageChannelBinder.this.getBeanFactory();
+			this.setBeanFactory(beanFactory);
+			this.partitionHandler = new PartitionHandler(beanFactory, evaluationContext, partitionSelector,
+					properties, properties.getNextModuleCount());
 			refreshChannelAdapters();
 		}
 
@@ -319,13 +323,13 @@ public class RedisMessageChannelBinder extends AbstractBinder<MessageChannel> {
 		protected void handleMessageInternal(Message<?> message) throws Exception {
 			MessageValues transformed = serializePayloadIfNecessary(message);
 
-			if (this.partitioningMetadata.isPartitionedModule()) {
-				transformed.put(PARTITION_HEADER, determinePartition(message, this.partitioningMetadata));
+			if (this.partitionHandler.isPartitionedModule()) {
+				transformed.put(PARTITION_HEADER, this.partitionHandler.determinePartition(message));
 			}
 
 			byte[] messageToSend = embeddedHeadersMessageConverter.embedHeaders(transformed,
 					RedisMessageChannelBinder.this.headersToMap);
-			
+
 			refreshChannelAdapters();
 			for (RedisQueueOutboundChannelAdapter adapter : adapters.values()) {
 				adapter.handleMessage((MessageBuilder.withPayload(messageToSend).copyHeaders(transformed).build()));