From efe6c87d48ff9e6c76767eb78a043e5d532128da Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 21 Jun 2023 13:50:10 -0400 Subject: [PATCH] Avoid StreamBridge double partition finding - 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 --- .../cloud/stream/function/StreamBridge.java | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index a38788767..01640f72a 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -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);