SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2245 StreamBridge partitioning fix

Browse Source 
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
This commit is contained in:
Soby Chacko
2021-11-24 13:11:56 -05:00
committed by Oleg Zhurakousky
parent 86832a9872
commit 321da8dcbc
1 changed files with 3 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -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;