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

Avoid StreamBridge double partition finding

Browse Source 
- In StreamBridge send method, there is a path, in which it
   tries to find the partition information twice, causing it
   to throw errors when using patition key expresson that
   involves the payload. This is because, the second time it
   tries to find the partition, the payload is already converted
   into byte[]. This second partition finding is unncessary.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2759
This commit is contained in:
Soby Chacko
2023-06-21 13:50:10 -04:00
parent 01efa99e9e
commit efe6c87d48
1 changed files with 8 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -38,7 +38,6 @@ import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor;
import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
@@ -246,12 +245,14 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
if (producerProperties != null && producerProperties.isPartitioned()) {
BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(destinationName);
((AbstractMessageChannel) messageChannel)
.addInterceptor(new DefaultPartitioningInterceptor(bindingProperties, this.applicationContext.getBeanFactory()));
}
// Commenting out the following block due to this issue: https://github.com/spring-cloud/spring-cloud-stream/issues/2759
// Once we confirm that there is no unknown regression, we will remove this block from here completely,
// since we already perform the partition finding algorithm once via StreamBridge#send.
// if (producerProperties != null && 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, true, binder);