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

GH-2815: StreamBridge partitioning fixes

Browse Source 
- When native encoding is used, StreamBridge is not setting
   partitioning interceptor which calls the PartitionHandler.
   This is a regression issue and adding the interceptor explicitly
   when native encoding is used.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2815
Resolves #2831
This commit is contained in:
Soby Chacko
2023-10-17 18:34:25 -04:00
committed by Oleg Zhurakousky
parent e56a7bd1cd
commit eebbe7a5bb
2 changed files with 28 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -179,6 +179,25 @@ public class StreamBridgeTests {
}
}
@Test
void ensurePartitioningWorksWhenNativeEncodingEnabled() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
EmptyConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.source=outputA;outputB",
"--spring.cloud.stream.bindings.outputA-out-0.producer.partition-count=3",
"--spring.cloud.stream.bindings.outputA-out-0.producer.use-native-encoding=true",
"--spring.cloud.stream.bindings.outputA-out-0.producer.partition-key-expression=headers['partitionKey']")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
streamBridge.send("outputA-out-0", MessageBuilder.withPayload("A").setHeader("partitionKey", "A").build());
streamBridge.send("outputA-out-0", MessageBuilder.withPayload("C").setHeader("partitionKey", "C").build());
OutputDestination output = context.getBean(OutputDestination.class);
assertThat(output.receive(1000, "outputA-out-0").getHeaders().containsKey("scst_partition")).isTrue();
assertThat(output.receive(1000, "outputA-out-0").getHeaders().containsKey("scst_partition")).isTrue();
}
}
@SuppressWarnings("rawtypes")
@Test
void test_2785() throws Exception {

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

@@ -38,7 +38,9 @@ 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;
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
import org.springframework.context.ConfigurableApplicationContext;
@@ -260,14 +262,13 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
// 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()));
// }
// since we already perform the partition finding algorithm once via StreamBridge#send we don't need to
// do the following, unless the conversion is handled natively on the middleware.
if (producerProperties != null && producerProperties.isPartitioned() && producerProperties.isUseNativeEncoding()) {
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);