From 7680e02cbb009b591cc9477fc4cb9d3fc6db71a2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 26 Feb 2025 17:47:23 -0500 Subject: [PATCH] GH-9866: Fix `FluxMessageChannel` for race condition on destroy 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`** --- .../channel/FluxMessageChannel.java | 28 +++++++++++-------- 1 file changed, 17 insertions(+), 11 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 7be166ca12..4f49fee71e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -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(); }