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

GH-2848: Partitioning issues in StreamBridge

Browse Source 
- When output-bindings property is explicitly provided and native-encoding
   is used, StreamBridge does not add the partitioning interceptor.
   This commit addresses this issue.

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2848
This commit is contained in:
Soby Chacko
2023-11-15 13:31:37 -05:00
committed by Oleg Zhurakousky
parent ba0924acc2
commit 06af5ceeb0
2 changed files with 35 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -181,12 +181,12 @@ class StreamBridgeTests {
}
}
// For more context on this test: https://github.com/spring-cloud/spring-cloud-stream/issues/2815
@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']")) {
@@ -200,6 +200,24 @@ class StreamBridgeTests {
}
}
// Fore more context on this test: https://github.com/spring-cloud/spring-cloud-stream/issues/2848
@Test
void ensurePartitioningWorksWhenNativeEncodingEnabledAndOutputBindingsExist() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
EmptyConfiguration.class)).web(WebApplicationType.NONE).run(
"--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.output-bindings=outputA-out-0",
"--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());
OutputDestination output = context.getBean(OutputDestination.class);
assertThat(output.receive(1000, "outputA-out-0").getHeaders().containsKey("scst_partition")).isTrue();
}
}
@SuppressWarnings("rawtypes")
@Test
void test_2785() throws Exception {

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

@@ -244,6 +244,9 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
if (this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
String[] consumerBindingNames = this.bindingService.getConsumerBindingNames();
if (messageChannel instanceof AbstractMessageChannel) {
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
}
if (ObjectUtils.containsElement(consumerBindingNames, destinationName)) { //GH-2563
logger.warn("You seem to be sending data to the input binding. It is not "
+ "recommended, since you are bypassing the binder and this the messaging system exposed by the binder.");
@@ -263,14 +266,8 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
// 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);
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
addGlobalChannelInterceptorProcessor((AbstractMessageChannel) messageChannel, destinationName);
this.bindingService.bindProducer(messageChannel, destinationName, true, binder);
if (StringUtils.hasText(binderName)) {
@@ -285,6 +282,16 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
return messageChannel;
}
private void addPartitioningInterceptorIfNeedBe(ProducerProperties producerProperties, String destinationName, AbstractMessageChannel messageChannel) {
// 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);
messageChannel
.addInterceptor(new DefaultPartitioningInterceptor(bindingProperties, this.applicationContext.getBeanFactory()));
}
}
private String resolveBinderTargetType(String channelName, String binderName, Class<?> bindableType, BinderFactory binderFactory) {
String binderConfigurationName = binderName != null ? binderName : this.bindingServiceProperties
.getBinder(channelName);
@@ -293,7 +300,7 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
return targetProtocol;
}
private void addInterceptors(AbstractMessageChannel messageChannel, String destinationName) {
private void addGlobalChannelInterceptorProcessor(AbstractMessageChannel messageChannel, String destinationName) {
final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor =
this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class);
globalChannelInterceptorProcessor.postProcessAfterInitialization(messageChannel, destinationName);