Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9866: Fix FluxMessageChannel for race condition on destroy

Browse Source 
Fixes: #9866
Issue link: https://github.com/spring-projects/spring-integration/issues/9866

Due to race condition between `sink.tryEmitNext()` and `sink.emitComplete()``,
there could be a situation when `onNext` signal slips after `onComplete`.
Appears on fast producers into this `FluxMessageChannel`.
That leads to error like `Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.`

* Fix `FluxMessageChannel` to check for `this.active` one more time just before `sink.tryEmitNext()` call
* Also mitigate a race condition with a `Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1))` in the `destroy()`
instead of `Sinks.EmitFailureHandler.FAIL_FAST`.
The `busyLooping()` would retry requested `onComplete()` signal for that specific error until success or timeout

**Auto-cherry-pick to `6.4.x` & `6.3.x`**
This commit is contained in:
Artem Bilan
2025-02-26 17:47:23 -05:00
parent d54d55653f
commit 7680e02cbb
1 changed files with 17 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2024 the original author or authors.
* Copyright 2015-2025 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -86,15 +86,21 @@ public class FluxMessageChannel extends AbstractMessageChannel
.setHeader(IntegrationMessageHeaderAccessor.REACTOR_CONTEXT, contextView)
.build();
}
return switch (this.sink.tryEmitNext(messageToEmit)) {
case OK -> true;
case FAIL_NON_SERIALIZED, FAIL_OVERFLOW -> false;
case FAIL_ZERO_SUBSCRIBER ->
throw new IllegalStateException("The [" + this + "] doesn't have subscribers to accept messages");
case FAIL_TERMINATED, FAIL_CANCELLED ->
throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: "
+ this.sink);
};
if (this.active) {
return switch (this.sink.tryEmitNext(messageToEmit)) {
case OK -> true;
case FAIL_NON_SERIALIZED, FAIL_OVERFLOW -> false;
case FAIL_ZERO_SUBSCRIBER ->
throw new IllegalStateException("The [" + this + "] doesn't have subscribers to accept messages");
case FAIL_TERMINATED, FAIL_CANCELLED ->
throw new IllegalStateException("Cannot emit messages into the cancelled or terminated sink: "
+ this.sink);
};
}
else {
return false;
}
}
@Override
@@ -171,7 +177,7 @@ public class FluxMessageChannel extends AbstractMessageChannel
public void destroy() {
this.active = false;
this.upstreamSubscriptions.dispose();
this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
this.sink.emitComplete(Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1)));
super.destroy();
}