From fa1b6257d5d4a723765d5c477f1682401e8a3a23 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 23 Jul 2018 12:56:02 -0400 Subject: [PATCH] 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 --- .../binder/AbstractMessageChannelBinder.java | 32 +++++++++++++++++-- .../cloud/stream/binder/PartitionHandler.java | 18 +++++++++-- .../binding/MessageConverterConfigurer.java | 14 ++++---- 3 files changed, 53 insertions(+), 11 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index edde7c0e2..810c139a7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -154,7 +154,7 @@ public abstract class AbstractMessageChannelBinder + * In order to be fully compliant, the {@link MessageHandler} of the binder must + * observe the following headers: + * + *

+ * + * @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. *

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 index bc177ab83..a6370677b 100644 --- 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 @@ -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. *

@@ -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) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 0e5e1a6f1..d8c2ff8e7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/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 } } } + }