Skip to content

The FluxMessageChannel.destroy() may fail with Spec. Rule 1.3 #9866

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Closed
artembilan opened this issue Feb 26, 2025 · 0 comments
Closed

The FluxMessageChannel.destroy() may fail with Spec. Rule 1.3 #9866

artembilan opened this issue Feb 26, 2025 · 0 comments
Assignees
@artembilan
Labels
for: backport-to-6.3.x for: backport-to-6.4.x in: core status: backported type: bug
Milestone
6.5.0-M3

Comments

@artembilan
Copy link
Member

artembilan commented Feb 26, 2025 •
edited
Loading

Due to race condition between this.sink.tryEmitNext() and this.sink.emitComplete(), there could be a situation when onNext signal slips after onComplete. Appears on fast producers into this FluxMessageChannel.

The error in logs looks like:

2025-02-26 17:39:51,315 WARN [main] [org.springframework.beans.factory.support.DisposableBeanAdapter] - Invocation of destroy method failed on bean with name 'org.springframework.integration.channel.FluxMessageChannel'
reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.
	at reactor.core.publisher.InternalManySink.emitComplete(InternalManySink.java:86) ~[reactor-core-3.7.3.jar:3.7.3]
	at org.springframework.integration.channel.FluxMessageChannel.destroy(FluxMessageChannel.java:174) ~[classes/:?]
	at org.springframework.beans.factory.support.DisposableBeanAdapter.destroy(DisposableBeanAdapter.java:211) ~[spring-beans-6.2.3.jar:6.2.3]

Right, the DisposableBeanAdapter hides an exception behind the warn message, but calling FluxMessageChannel.destroy() may lead to failures in the target application.
For example, when beans are destroyed manually, not via BeanFactory.
@artembilan artembilan added this to the 6.5.0-M3 milestone Feb 26, 2025
@artembilan artembilan self-assigned this Feb 26, 2025
@artembilan artembilan changed the title The FluxMessageChannel.destroy() may fail with Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially. The FluxMessageChannel.destroy() may fail with Spec. Rule 1.3 Feb 26, 2025
spring-builds pushed a commit that referenced this issue Feb 26, 2025
@artembilan @spring-builds
GH-9866: Fix FluxMessageChannel for race condition on destroy
33894dc 
Fixes: #9866
Issue link: #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

(cherry picked from commit 7680e02)
spring-builds pushed a commit that referenced this issue Feb 26, 2025
@artembilan @spring-builds
GH-9866: Fix FluxMessageChannel for race condition on destroy
d2e1442 
Fixes: #9866
Issue link: #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

(cherry picked from commit 7680e02)
This was referenced Feb 26, 2025
Closed
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees

@artembilan artembilan

Labels
for: backport-to-6.3.x for: backport-to-6.4.x in: core status: backported type: bug
Projects
None yet
Milestone
6.5.0-M3
Development

No branches or pull requests
2 participants
@artembilan @spring-builds