From 06af5ceeb01465b198955188d3daa6c90016d2be Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 15 Nov 2023 13:31:37 -0500 Subject: [PATCH] GH-2848: Partitioning issues in StreamBridge - 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 --- .../stream/function/StreamBridgeTests.java | 20 ++++++++++++++- .../cloud/stream/function/StreamBridge.java | 25 ++++++++++++------- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java index 80f5a4ef8..bee6acfba 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java @@ -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 { 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 df85dc557..9e44bb80d 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 @@ -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);