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

Remove redundant delaySubscription from FunctionConfiguration

Browse Source 
Related to: https://github.com/spring-projects/spring-integration/issues/9362

After the fix in Spring Integration: bdcb856a90
we ended up in a deadlock situation with a `beginPublishingTrigger` in the `FunctionConfiguration`
used for the `delaySubscription()` on an original `Publisher`.
The `FluxMessageChannel` uses its own `delaySubscription()` until the channel has its subscribers.
Apparently the logic before was with errors, so the `FluxMessageChannel` was marked as active
even if its subscriber is not ready yet, leading to famous `Dispatcher does not have subscribers` error.
So, looks like this `beginPublishingTrigger` was introduced back in days in Spring Cloud Stream
to mitigate that situation until we really emit a `BindingCreatedEvent`.

The deadlock (and the flaw, respectively) is with the `setupBindingTrigger()` method implementation
where `FluxMessageChannel` now "really" delays a subscription to the provided `Publisher`,
therefore not triggering that `Mono.create()` fulfilment immediately.
The `BindingCreatedEvent` arrives earlier, than we have a subscriber on the channel,
but `triggerRef.get()` is `null`, so we don't `success()` it and in the end don't subscribe
to an original `Publisher` since `delaySubscription()` on it is never completed.

Since `FunctionConfiguration` fully relies on `IntegrationFlow.from(Publisher)`,
which ends up with the mentioned  `FluxMessageChannel.subscribeTo()` and its own `delaySubscription()`
(which, in turn, apparently fixed now), we don't need our own `delaySubscription()` any more.
Therefore the fix in this PR is to propose to remove `beginPublishingTrigger` logic altogether.
This commit is contained in:
Artem Bilan
2024-07-26 15:57:12 -04:00
parent f356599503
commit 3b5bb2e0d9
1 changed files with 6 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -40,7 +40,6 @@ import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.util.function.Tuples;
import org.springframework.beans.BeansException;
@@ -65,7 +64,6 @@ import org.springframework.cloud.function.context.config.RoutingFunction;
import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderHeaders;
import org.springframework.cloud.stream.binder.BindingCreatedEvent;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.PartitionHandler;
import org.springframework.cloud.stream.binder.ProducerProperties;
@@ -129,6 +127,7 @@ import org.springframework.util.StringUtils;
* @author Byungjun You
* @author Ivan Shapoval
* @author Patrik Péter Süli
* @author Artem Bilan
* @since 2.1
*/
@Lazy(false)
@@ -224,8 +223,6 @@ public class FunctionConfiguration {
functionWrapper = functionCatalog.lookup(proxyFactory.getFunctionDefinition(), contentTypes.toArray(new String[0]));
}
Publisher<Object> beginPublishingTrigger = setupBindingTrigger(context);
if (!functionProperties.isComposeFrom() && !functionProperties.isComposeTo()) {
String integrationFlowName = proxyFactory.getFunctionDefinition() + "_integrationflow";
@@ -239,7 +236,7 @@ public class FunctionConfiguration {
if (functionWrapper != null) {
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(functionWrapper, context, producerProperties),
beginPublishingTrigger, pollable, context, taskScheduler, producerProperties, outputName)
pollable, context, taskScheduler, producerProperties, outputName)
.route(Message.class, message -> {
if (message.getHeaders().get("spring.cloud.stream.sendto.destination") != null) {
String destinationName = (String) message.getHeaders().get("spring.cloud.stream.sendto.destination");
@@ -253,7 +250,7 @@ public class FunctionConfiguration {
}
else {
IntegrationFlow integrationFlow = integrationFlowFromProvidedSupplier(new PartitionAwareFunctionWrapper(supplier, context, producerProperties),
beginPublishingTrigger, pollable, context, taskScheduler, producerProperties, outputName)
pollable, context, taskScheduler, producerProperties, outputName)
.channel(c -> c.direct())
.fluxTransform((Function<? super Flux<Message<Object>>, ? extends Publisher<Object>>) function)
.route(Message.class, message -> {
@@ -274,26 +271,9 @@ public class FunctionConfiguration {
};
}
/*
* Creates a publishing trigger to ensure Supplier does not begin publishing until binding is created
*/
private Publisher<Object> setupBindingTrigger(GenericApplicationContext context) {
AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
Publisher<Object> beginPublishingTrigger = Mono.create(triggerRef::set);
context.addApplicationListener(event -> {
if (event instanceof BindingCreatedEvent) {
if (triggerRef.get() != null) {
triggerRef.get().success();
}
}
});
return beginPublishingTrigger;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier<?> supplier,
Publisher<Object> beginPublishingTrigger, PollableBean pollable, GenericApplicationContext context,
PollableBean pollable, GenericApplicationContext context,
TaskScheduler taskScheduler, ProducerProperties producerProperties, String bindingName) {
IntegrationFlowBuilder integrationFlowBuilder;
@@ -309,8 +289,8 @@ public class FunctionConfiguration {
if (pollable == null && reactive) {
Publisher publisher = (Publisher) supplier.get();
publisher = publisher instanceof Mono
? ((Mono) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary)
: ((Flux) publisher).delaySubscription(beginPublishingTrigger).map(this::wrapToMessageIfNecessary);
? ((Mono) publisher).map(this::wrapToMessageIfNecessary)
: ((Flux) publisher).map(this::wrapToMessageIfNecessary);
integrationFlowBuilder = IntegrationFlow.from(publisher);