From d34aafb87beeaec2e276ea3709dba825ce261eeb Mon Sep 17 00:00:00 2001 From: Patrick Peralta Date: Wed, 17 Feb 2016 11:26:39 -0500 Subject: [PATCH] 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 --- .../kafka/KafkaMessageChannelBinder.java | 25 +- .../rabbit/RabbitMessageChannelBinder.java | 14 +- .../redis/RedisMessageChannelBinder.java | 18 +- .../cloud/stream/binder/AbstractBinder.java | 141 +---------- .../cloud/stream/binder/PartitionHandler.java | 234 ++++++++++++++++++ 5 files changed, 268 insertions(+), 164 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index ffcb76f23..c0117c347 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 { private static final boolean DEFAULT_SYNC_PRODUCER = false; - private static final StartOffset DEFAULT_START_OFFSET = StartOffset.latest; - - private RetryOperations retryOperations; - - private Map> topicsInUse = new HashMap<>(); - protected static final Set PRODUCER_COMPRESSION_PROPERTIES = new HashSet( Arrays.asList(new String[] { KafkaMessageChannelBinder.COMPRESSION_CODEC, @@ -207,6 +203,10 @@ public class KafkaMessageChannelBinder extends AbstractBinder { .addAll(PRODUCER_COMPRESSION_PROPERTIES) .build(); + private RetryOperations retryOperations; + + private final Map> topicsInUse = new HashMap<>(); + private final EmbeddedHeadersMessageConverter embeddedHeadersMessageConverter = new EmbeddedHeadersMessageConverter(); @@ -830,8 +830,6 @@ public class KafkaMessageChannelBinder extends AbstractBinder { 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 { private final ProducerConfiguration producerConfiguration; + private final PartitionHandler partitionHandler; private SendingHandler(String topicName, KafkaPropertiesAccessor properties, int numberOfPartitions, ProducerConfiguration 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; diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index a9b983175..3b45c2de8 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/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 { 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 { 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())); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java index bd54d77d2..f7ad7aa7c 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/java/org/springframework/cloud/stream/binder/redis/RedisMessageChannelBinder.java +++ b/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 { private final String bindingName; - private final PartitioningMetadata partitioningMetadata; - private final RedisPropertiesAccessor accessor; private final Map 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 { 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())); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java index c45af9551..b012963be 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java @@ -45,7 +45,6 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.core.serializer.support.SerializationFailedException; import org.springframework.expression.EvaluationContext; -import org.springframework.expression.Expression; import org.springframework.integration.codec.Codec; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.messaging.Message; @@ -147,7 +146,7 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init protected volatile EvaluationContext evaluationContext; - private volatile PartitionSelectorStrategy partitionSelector = new DefaultPartitionSelector(); + protected volatile PartitionSelectorStrategy partitionSelector; protected volatile long defaultBackOffInitialInterval = DEFAULT_BACKOFF_INITIAL_INTERVAL; @@ -449,98 +448,6 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init } } - /** - * Determine the partition to which to send this message. If a partition key extractor class is provided, it is - * invoked to determine the key. Otherwise, the partition key expression is evaluated to obtain the key value. If a - * partition selector class is provided, it will be invoked to determine the partition. Otherwise, if the partition - * expression is not null, it is evaluated against the key and is expected to return an integer to which the modulo - * function will be applied, using the partitionCount as the divisor. If no partition expression is provided, the - * key will be passed to the binder partition strategy along with the partitionCount. The default partition - * strategy - * uses {@code key.hashCode()}, and the result will be the mod of that value. - * @param message the message. - * @param meta the partitioning metadata. - * @return the partition. - */ - protected int determinePartition(Message message, PartitioningMetadata meta) { - Object key = null; - if (StringUtils.hasText(meta.partitionKeyExtractorClass)) { - key = invokeExtractor(meta.partitionKeyExtractorClass, message); - } - else if (meta.partitionKeyExpression != null) { - key = meta.partitionKeyExpression.getValue(this.evaluationContext, message); - } - Assert.notNull(key, "Partition key cannot be null"); - int partition; - if (StringUtils.hasText(meta.partitionSelectorClass)) { - partition = invokePartitionSelector(meta.partitionSelectorClass, key, meta.partitionCount); - } - else if (meta.partitionSelectorExpression != null) { - partition = meta.partitionSelectorExpression.getValue(this.evaluationContext, key, Integer.class); - } - else { - partition = this.partitionSelector.selectPartition(key, meta.partitionCount); - } - partition = partition % meta.partitionCount; - if (partition < 0) { // protection in case a user selector returns a negative. - partition = Math.abs(partition); - } - return partition; - } - - private Object invokeExtractor(String partitionKeyExtractorClassName, Message message) { - if (this.applicationContext.containsBean(partitionKeyExtractorClassName)) { - return this.applicationContext.getBean(partitionKeyExtractorClassName, PartitionKeyExtractorStrategy.class) - .extractKey(message); - } - Class clazz; - try { - clazz = ClassUtils.forName(partitionKeyExtractorClassName, this.applicationContext.getClassLoader()); - } - catch (Exception e) { - this.logger.error("Failed to load key extractor", e); - throw new BinderException("Failed to load key extractor: " + partitionKeyExtractorClassName, e); - } - try { - Object extractor = clazz.newInstance(); - Assert.isInstanceOf(PartitionKeyExtractorStrategy.class, extractor); - this.applicationContext.getBeanFactory().registerSingleton(partitionKeyExtractorClassName, extractor); - this.applicationContext.getBeanFactory().initializeBean(extractor, partitionKeyExtractorClassName); - return ((PartitionKeyExtractorStrategy) extractor).extractKey(message); - } - catch (Exception e) { - this.logger.error("Failed to instantiate key extractor", e); - throw new BinderException("Failed to instantiate key extractor: " + partitionKeyExtractorClassName, e); - } - } - - private int invokePartitionSelector(String partitionSelectorClassName, Object key, int partitionCount) { - if (this.applicationContext.containsBean(partitionSelectorClassName)) { - return this.applicationContext.getBean(partitionSelectorClassName, PartitionSelectorStrategy.class) - .selectPartition(key, partitionCount); - } - Class clazz; - try { - clazz = ClassUtils.forName(partitionSelectorClassName, this.applicationContext.getClassLoader()); - } - catch (Exception e) { - this.logger.error("Failed to load partition selector", e); - throw new BinderException("Failed to load partition selector: " + partitionSelectorClassName, e); - } - try { - Object extractor = clazz.newInstance(); - Assert.isInstanceOf(PartitionKeyExtractorStrategy.class, extractor); - this.applicationContext.getBeanFactory().registerSingleton(partitionSelectorClassName, extractor); - this.applicationContext.getBeanFactory().initializeBean(extractor, partitionSelectorClassName); - return ((PartitionSelectorStrategy) extractor).selectPartition(key, partitionCount); - } - catch (Exception e) { - this.logger.error("Failed to instantiate partition selector", e); - throw new BinderException("Failed to instantiate partition selector: " + partitionSelectorClassName, - e); - } - } - /** * Validate the provided deployment properties for the consumer against those supported by this binder * implementation. @@ -620,52 +527,6 @@ public abstract class AbstractBinder implements ApplicationContextAware, Init } } - /** - * Default partition strategy; only works on keys with "real" hash codes, such as String. Caller now always applies - * modulo so no need to do so here. - */ - private class DefaultPartitionSelector implements PartitionSelectorStrategy { - - @Override - public int selectPartition(Object key, int partitionCount) { - int hashCode = key.hashCode(); - if (hashCode == Integer.MIN_VALUE) { - hashCode = 0; - } - return Math.abs(hashCode); - } - - } - - protected static class PartitioningMetadata { - - private final String partitionKeyExtractorClass; - - private final Expression partitionKeyExpression; - - private final String partitionSelectorClass; - - private final Expression partitionSelectorExpression; - - private final int partitionCount; - - public PartitioningMetadata(DefaultBindingPropertiesAccessor properties, int partitionCount) { - this.partitionCount = partitionCount; - this.partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass(); - this.partitionKeyExpression = properties.getPartitionKeyExpression(); - this.partitionSelectorClass = properties.getPartitionSelectorClass(); - this.partitionSelectorExpression = properties.getPartitionSelectorExpression(); - } - - public boolean isPartitionedModule() { - return StringUtils.hasText(this.partitionKeyExtractorClass) || this.partitionKeyExpression != null; - } - - public int getPartitionCount() { - return this.partitionCount; - } - - } /** * Handles representing any java class as a {@link MimeType}. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java new file mode 100644 index 000000000..43990ec40 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/PartitionHandler.java @@ -0,0 +1,234 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.messaging.Message; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.StringUtils; + +/** + * Utility class to determine if a binding is configured for partitioning + * (based on the binder properties provided in the constructor) and + * what partition a message should be delivered to. + * + * @author Patrick Peralta + * @author David Turanski + * @author Gary Russell + * @author Ilayaperumal Gopinathan + * @author Mark Fisher + * @author Marius Bogoevici + */ +public class PartitionHandler { + + private final ConfigurableListableBeanFactory beanFactory; + + private final EvaluationContext evaluationContext; + + private final PartitionSelectorStrategy partitionSelector; + + private final PartitioningMetadata metadata; + + + /** + * Construct a {@code PartitionHandler}. + * + * @param beanFactory bean factory for binder + * @param evaluationContext evaluation context for binder + * @param partitionSelector configured partition selector; may be {@code null} + * @param properties binder properties + * @param partitionCount number of partitions configured for binder + */ + public PartitionHandler(ConfigurableListableBeanFactory beanFactory, + EvaluationContext evaluationContext, + PartitionSelectorStrategy partitionSelector, + DefaultBindingPropertiesAccessor properties, int partitionCount) { + this.beanFactory = beanFactory; + this.evaluationContext = evaluationContext; + this.partitionSelector = partitionSelector == null + ? new DefaultPartitionSelector() + : partitionSelector; + this.metadata = new PartitioningMetadata(properties, partitionCount); + } + + /** + * Return {@code true} if the binder properties provided indicate + * that this binder is configured for partitioning. + * + * @return true if partitioning is enabled + */ + public boolean isPartitionedModule() { + return this.metadata.isPartitionedModule(); + } + + /** + * Determine the partition to which to send this message. + *

+ * If a partition key extractor class is provided, it is invoked to determine + * the key. Otherwise, the partition key expression is evaluated to obtain the + * key value. + *

+ * If a partition selector class is provided, it will be invoked to determine the + * partition. Otherwise, if the partition expression is not null, it is evaluated + * against the key and is expected to return an integer to which the modulo + * function will be applied, using the {@code partitionCount} as the divisor. If no + * partition expression is provided, the key will be passed to the binder + * partition strategy along with the {@code partitionCount}. The default partition + * strategy uses {@code key.hashCode()}, and the result will be the mod of that value. + * + * @param message the message. + * @return the partition + */ + public int determinePartition(Message message) { + Object key = extractKey(message); + + int partition; + if (this.metadata.hasSelectorClass()) { + partition = invokePartitionSelector(key); + } + else if (this.metadata.hasSelectorExpression()) { + partition = this.metadata.partitionSelectorExpression.getValue( + this.evaluationContext, key, Integer.class); + } + else { + partition = this.partitionSelector.selectPartition(key, metadata.partitionCount); + } + // protection in case a user selector returns a negative. + return Math.abs(partition % metadata.partitionCount); + } + + private Object extractKey(Message message) { + Object key = null; + if (this.metadata.hasKeyExtractorClass()) { + key = invokeKeyExtractor(message); + } + else if (this.metadata.hasKeyExpression()) { + key = this.metadata.partitionKeyExpression.getValue(this.evaluationContext, message); + } + Assert.notNull(key, "Partition key cannot be null"); + + return key; + } + + private Object invokeKeyExtractor(Message message) { + PartitionKeyExtractorStrategy strategy = getBean( + metadata.partitionKeyExtractorClass, + PartitionKeyExtractorStrategy.class); + return strategy.extractKey(message); + } + + private int invokePartitionSelector(Object key) { + PartitionSelectorStrategy strategy = getBean( + metadata.partitionSelectorClass, + PartitionSelectorStrategy.class); + return strategy.selectPartition(key, metadata.partitionCount); + } + + private T getBean(String className, Class type) { + if (this.beanFactory.containsBean(className)) { + return this.beanFactory.getBean(className, type); + } + else { + synchronized (this) { + if (this.beanFactory.containsBean(className)) { + return this.beanFactory.getBean(className, type); + } + Class clazz; + try { + clazz = ClassUtils.forName(className, this.beanFactory.getBeanClassLoader()); + } + catch (Exception e) { + throw new BinderException("Failed to load class: " + className, e); + } + try { + @SuppressWarnings("unchecked") + T object = (T) clazz.newInstance(); + Assert.isInstanceOf(type, object); + this.beanFactory.registerSingleton(className, object); + this.beanFactory.initializeBean(object, className); + return object; + } + catch (Exception e) { + throw new BinderException("Failed to instantiate class: " + className, e); + } + } + } + } + + /** + * Default partition strategy; only works on keys with "real" hash codes, + * such as String. Caller now always applies modulo so no need to do so here. + */ + private static class DefaultPartitionSelector implements PartitionSelectorStrategy { + + @Override + public int selectPartition(Object key, int partitionCount) { + int hashCode = key.hashCode(); + if (hashCode == Integer.MIN_VALUE) { + hashCode = 0; + } + return Math.abs(hashCode); + } + + } + + private static class PartitioningMetadata { + + private final String partitionKeyExtractorClass; + + private final Expression partitionKeyExpression; + + private final String partitionSelectorClass; + + private final Expression partitionSelectorExpression; + + private final int partitionCount; + + public PartitioningMetadata(DefaultBindingPropertiesAccessor properties, int partitionCount) { + this.partitionCount = partitionCount; + this.partitionKeyExtractorClass = properties.getPartitionKeyExtractorClass(); + this.partitionKeyExpression = properties.getPartitionKeyExpression(); + this.partitionSelectorClass = properties.getPartitionSelectorClass(); + this.partitionSelectorExpression = properties.getPartitionSelectorExpression(); + } + + public boolean isPartitionedModule() { + return StringUtils.hasText(this.partitionKeyExtractorClass) || this.partitionKeyExpression != null; + } + + public boolean hasSelectorClass() { + return StringUtils.hasText(this.partitionSelectorClass); + } + + public boolean hasKeyExtractorClass() { + return StringUtils.hasText(this.partitionKeyExtractorClass); + } + + public boolean hasSelectorExpression() { + return partitionSelectorExpression != null; + } + + public boolean hasKeyExpression() { + return partitionKeyExpression != null; + } + + } + +}