From 321da8dcbc5bc9809aef6a5c649122502eabc82e Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 24 Nov 2021 13:11:56 -0500 Subject: [PATCH] GH-2245 StreamBridge partitioning fix Fix the order in applying partitioning interceptor in StreamBridge. The interceptor must be added before the call for binding the producer. Related to resolving https://github.com/spring-cloud/spring-cloud-stream/issues/2245 Specifically for this: https://github.com/spring-cloud/spring-cloud-stream/issues/2245#issuecomment-977663452 --- .../springframework/cloud/stream/function/StreamBridge.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index 01ddedc87..bab3aa32d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -271,14 +271,15 @@ public final class StreamBridge implements SmartInitializingSingleton { binder = binderFactory.getBinder(binderName, messageChannel.getClass()); } - this.bindingService.bindProducer(messageChannel, destinationName, false, binder); - this.channelCache.put(destinationName, messageChannel); if (producerProperties.isPartitioned()) { BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(destinationName); ((AbstractMessageChannel) messageChannel) .addInterceptor(new DefaultPartitioningInterceptor(bindingProperties, this.applicationContext.getBeanFactory())); } this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName); + + this.bindingService.bindProducer(messageChannel, destinationName, false, binder); + this.channelCache.put(destinationName, messageChannel); } return messageChannel;