From e39ca9691fe10f9f1866f14356b50fbdf3cac836 Mon Sep 17 00:00:00 2001 From: arimu1 <19286898+arimu1@users.noreply.github.com> Date: Fri, 14 Aug 2026 07:18:40 +0700 Subject: [PATCH 1/3] Fix message loss on stop() for PollableChannel reactive consumer Rescue in-flight messages via Reactor discard hooks on the existing Mono.create poll loop: doOnDiscard for the CANCELLED path and a sequence-local onNextDropped context hook for TERMINATED. PollableChannel sources are best-effort re-queued with non-blocking send(message, 0), falling back to nack when re-queue fails; other MessageSource types nack via AcknowledgmentCallback when present. Replace probabilistic soak loops with a deterministic test that blocks in receive() until after stop() cancels the subscription. Fixes gh-11262 Signed-off-by: arimu1 <19286898+arimu1@users.noreply.github.com> --- .../util/IntegrationReactiveUtils.java | 80 +++++++++++++++++-- .../IntegrationReactiveUtilsTests.java | 45 +++++++++++ .../ReactiveStreamsConsumerTests.java | 68 ++++++++++++++++ 3 files changed, 187 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index 26654e45c61..2a26dbc0ac3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -18,6 +18,7 @@ import java.time.Duration; import java.util.concurrent.locks.LockSupport; +import java.util.function.Consumer; import io.micrometer.context.ContextSnapshotFactory; import org.apache.commons.logging.Log; @@ -49,6 +50,7 @@ * Utilities for adapting integration components to/from reactive types. * * @author Artem Bilan + * @author Fardan An * * @since 5.3 */ @@ -56,6 +58,12 @@ public final class IntegrationReactiveUtils { private static final Log LOGGER = LogFactory.getLog(IntegrationReactiveUtils.class); + /** + * Reactor context key for a sequence-local {@code onNextDropped} hook. + * @see reactor.core.publisher.Operators#onNextDropped(Object, reactor.util.context.Context) + */ + private static final String REACTOR_ON_NEXT_DROPPED_CONTEXT_KEY = "reactor.onNextDropped.local"; + /** * The subscriber context entry for {@link Flux#delayElements} * from the {@link Mono#repeatWhenEmpty(java.util.function.Function)}. @@ -123,15 +131,29 @@ public static ContextView captureReactorContext() { * If a produced message has an * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#ACKNOWLEDGMENT_CALLBACK} header * it is ack'ed in the {@link Mono#doOnSuccess} and nack'ed in the {@link Mono#doOnError}. + *

+ * When the subscription is cancelled while a message is in flight (already received from the + * {@link MessageSource} but not yet delivered to a downstream subscriber), Reactor routes the + * value through discard hooks which rescue it here: for a {@link PollableChannel} the message is + * best-effort re-queued via non-blocking {@link PollableChannel#send(Message, long)}; for other + * sources an {@link org.springframework.integration.acks.AcknowledgmentCallback} is nack'd when + * present. * @param messageSource the {@link MessageSource} to adapt. * @param the expected payload type. * @return a {@link Flux} which pulls messages from the {@link MessageSource} on demand. */ @SuppressWarnings("NullAway") public static Flux> messageSourceToFlux(MessageSource messageSource) { - return Mono. - >create(monoSink -> - monoSink.onRequest(value -> monoSink.success(messageSource.receive()))) + Consumer undeliveredConsumer = createUndeliveredConsumer(messageSource); + + return Mono.>create(monoSink -> monoSink.onRequest(request -> { + try { + monoSink.success(messageSource.receive()); + } + catch (Exception ex) { + monoSink.error(ex); + } + })) .doOnSuccess((message) -> { if (message != null) { AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); @@ -152,6 +174,8 @@ public static Flux> messageSourceToFlux(MessageSource messageS Mono.delay(ctx.getOrDefault(DELAY_WHEN_EMPTY_KEY, DEFAULT_DELAY_WHEN_EMPTY))))) .repeat() + .doOnDiscard(Message.class, message -> handleUndeliveredMessage(messageSource, message)) + .contextWrite(ctx -> ctx.put(REACTOR_ON_NEXT_DROPPED_CONTEXT_KEY, undeliveredConsumer)) .retryWhen(Retry.indefinitely().filter(MessagingException.class::isInstance)); } @@ -161,7 +185,7 @@ public static Flux> messageSourceToFlux(MessageSource messageS * is returned as is because it is already a {@link Publisher}; * - a {@link SubscribableChannel} is subscribed with a {@link MessageHandler} * for the {@link Sinks.Many#tryEmitNext(Object)} which is returned from this method; - * - a {@link PollableChannel} is wrapped into a {@link MessageSource} lambda and reuses + * - a {@link PollableChannel} is wrapped into a {@link PollableChannelMessageSource} and reuses * {@link #messageSourceToFlux(MessageSource)}. * @param messageChannel the {@link MessageChannel} to adapt. * @param the expected payload type. @@ -175,8 +199,8 @@ public static Flux> messageChannelToFlux(MessageChannel messageCh else if (messageChannel instanceof SubscribableChannel) { return adaptSubscribableChannelToPublisher((SubscribableChannel) messageChannel); } - else if (messageChannel instanceof PollableChannel) { - return messageSourceToFlux(() -> (Message) ((PollableChannel) messageChannel).receive(0)); + else if (messageChannel instanceof PollableChannel pollableChannel) { + return messageSourceToFlux(new PollableChannelMessageSource<>(pollableChannel)); } else { throw new IllegalArgumentException("The 'messageChannel' must be an instance of Publisher, " + @@ -184,6 +208,50 @@ else if (messageChannel instanceof PollableChannel) { } } + private static Consumer createUndeliveredConsumer(MessageSource messageSource) { + return value -> { + if (value instanceof Message message) { + handleUndeliveredMessage(messageSource, message); + } + }; + } + + private static void handleUndeliveredMessage(MessageSource messageSource, @Nullable Message message) { + if (message == null) { + return; + } + if (messageSource instanceof PollableChannelMessageSource pollableChannelMessageSource) { + pollableChannelMessageSource.returnMessage(message); + } + else { + AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); + } + } + + private static final class PollableChannelMessageSource implements MessageSource { + + private final PollableChannel channel; + + PollableChannelMessageSource(PollableChannel channel) { + this.channel = channel; + } + + @Override + @SuppressWarnings("unchecked") + public @Nullable Message receive() { + return (Message) this.channel.receive(0); + } + + void returnMessage(Message message) { + if (!this.channel.send(message, 0)) { + LOGGER.warn("Failed to return undelivered message to pollable channel [" + this.channel + + "]; nacking instead"); + AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); + } + } + + } + @SuppressWarnings("unchecked") private static Flux> adaptSubscribableChannelToPublisher(SubscribableChannel inputChannel) { return Flux.defer(() -> { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java index 4717bf1d930..c075de30a9d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java @@ -19,8 +19,10 @@ import java.time.Duration; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Test; @@ -32,16 +34,22 @@ import reactor.test.StepVerifier; import reactor.util.concurrent.Queues; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.core.MessageSource; import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.messaging.Message; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; /** * @author Sergei Egorov * @author Artem Bilan + * @author Fardan An * * @since 5.1.9 */ @@ -150,4 +158,41 @@ void testRetryOnMessagingExceptionOnly() { assertThat(finalException.get()).hasMessage("non-retryable RuntimeException"); } + @Test + void undeliveredMessageNackedWhenCancelledDuringBlockingReceive() throws InterruptedException { + CountDownLatch receiveBlocked = new CountDownLatch(1); + AtomicBoolean releaseReceive = new AtomicBoolean(); + AtomicReference ackStatus = new AtomicReference<>(); + AtomicReference> delivered = new AtomicReference<>(); + + Message testMessage = MessageBuilder.withPayload("test") + .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, + (AcknowledgmentCallback) ackStatus::set) + .build(); + + MessageSource blockingSource = () -> { + receiveBlocked.countDown(); + while (!releaseReceive.get()) { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); // NOSONAR busy wait + } + @SuppressWarnings("unchecked") + Message message = (Message) testMessage; + return message; + }; + + Disposable subscription = IntegrationReactiveUtils.messageSourceToFlux(blockingSource) + .subscribe(delivered::set); + + assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); + + subscription.dispose(); + releaseReceive.set(true); + + await().atMost(Duration.ofSeconds(5)) + .until(() -> ackStatus.get() == AcknowledgmentCallback.Status.REJECT); + + assertThat(delivered.get()).isNull(); + assertThat(ackStatus.get()).isEqualTo(AcknowledgmentCallback.Status.REJECT); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 21f92113828..75fb791cea3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -23,7 +23,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.LockSupport; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -65,6 +67,7 @@ /** * @author Artem Bilan + * @author Fardan An * * @since 5.0 */ @@ -248,6 +251,71 @@ public void testReactiveStreamsConsumerPollableChannel() throws InterruptedExcep reactiveConsumer.stop(); } + @Test + @SuppressWarnings("unchecked") + public void messageNotLostWhenStopDuringBlockingReceive() throws InterruptedException { + CountDownLatch receiveBlocked = new CountDownLatch(1); + AtomicBoolean releaseReceive = new AtomicBoolean(); + + QueueChannel testChannel = new QueueChannel(1) { + + @Override + public Message receive(long timeout) { + Message message = super.receive(timeout); + if (message != null) { + receiveBlocked.countDown(); + while (!releaseReceive.get()) { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); // NOSONAR busy wait + } + } + return message; + } + + }; + + Subscriber> testSubscriber = (Subscriber>) Mockito.mock(Subscriber.class); + AtomicReference> delivered = new AtomicReference<>(); + + willAnswer(i -> { + delivered.set(i.getArgument(0)); + return null; + }).given(testSubscriber).onNext(any(Message.class)); + + ReactiveStreamsConsumer reactiveConsumer = new ReactiveStreamsConsumer(testChannel, testSubscriber); + reactiveConsumer.setBeanFactory(TEST_INTEGRATION_CONTEXT); + reactiveConsumer.afterPropertiesSet(); + reactiveConsumer.start(); + + Message testMessage = new GenericMessage<>("test"); + testChannel.send(testMessage); + + ArgumentCaptor captor = ArgumentCaptor.forClass(Subscription.class); + verify(testSubscriber).onSubscribe(captor.capture()); + captor.getValue().request(1); + + assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(testChannel.getQueueSize()).isZero(); + + reactiveConsumer.stop(); + releaseReceive.set(true); + + await().atMost(Duration.ofSeconds(5)) + .until(() -> delivered.get() != null || testChannel.getQueueSize() > 0); + + boolean wasDelivered = delivered.get() != null; + boolean wasRequeued = testChannel.getQueueSize() > 0; + assertThat(wasDelivered ^ wasRequeued).isTrue(); + + if (wasDelivered) { + assertThat(delivered.get()).isSameAs(testMessage); + } + else { + assertThat(testChannel.receive(0)).isSameAs(testMessage); + } + + reactiveConsumer.stop(); + } + @Test public void testReactiveStreamsConsumerViaConsumerEndpointFactoryBean() throws Exception { FluxMessageChannel testChannel = new FluxMessageChannel(); From dff9f151d5be74fffa93eb5b6211dc444ea6d8f0 Mon Sep 17 00:00:00 2001 From: Fardan An <19286898+arimu1@users.noreply.github.com> Date: Sat, 15 Aug 2026 14:43:21 +0700 Subject: [PATCH 2/3] GH-11262: Do not nack cancelled in-flight source messages Fixes: https://github.com/spring-projects/spring-integration/issues/11262 Cancellation during receive must not REJECT Kafka/AMQP deliveries. Those sources recover when left unacknowledged; only a PollableChannel needs a best-effort re-queue. * Re-queue PollableChannel only; never autoNack on discard * Move autoAck to Flux.doOnNext (delivery to this flux) * Drop the private reactor.onNextDropped.local hook * Restore receive() without a wrapping try/catch * Release blocking-receive test latches in finally Signed-off-by: Fardan An <19286898+arimu1@users.noreply.github.com> --- .../util/IntegrationReactiveUtils.java | 68 ++++++------------- .../IntegrationReactiveUtilsTests.java | 36 ++++++---- .../ReactiveStreamsConsumerTests.java | 27 +++----- 3 files changed, 53 insertions(+), 78 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index 2a26dbc0ac3..41ecd75e320 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -18,7 +18,6 @@ import java.time.Duration; import java.util.concurrent.locks.LockSupport; -import java.util.function.Consumer; import io.micrometer.context.ContextSnapshotFactory; import org.apache.commons.logging.Log; @@ -58,12 +57,6 @@ public final class IntegrationReactiveUtils { private static final Log LOGGER = LogFactory.getLog(IntegrationReactiveUtils.class); - /** - * Reactor context key for a sequence-local {@code onNextDropped} hook. - * @see reactor.core.publisher.Operators#onNextDropped(Object, reactor.util.context.Context) - */ - private static final String REACTOR_ON_NEXT_DROPPED_CONTEXT_KEY = "reactor.onNextDropped.local"; - /** * The subscriber context entry for {@link Flux#delayElements} * from the {@link Mono#repeatWhenEmpty(java.util.function.Function)}. @@ -130,35 +123,27 @@ public static ContextView captureReactorContext() { * or falls back to 1-second duration. * If a produced message has an * {@link org.springframework.integration.IntegrationMessageHeaderAccessor#ACKNOWLEDGMENT_CALLBACK} header - * it is ack'ed in the {@link Mono#doOnSuccess} and nack'ed in the {@link Mono#doOnError}. + * it is ack'ed in {@link Flux#doOnNext} when this flux emits to its subscriber, and nack'ed in + * the {@link Mono#doOnError}. *

- * When the subscription is cancelled while a message is in flight (already received from the - * {@link MessageSource} but not yet delivered to a downstream subscriber), Reactor routes the - * value through discard hooks which rescue it here: for a {@link PollableChannel} the message is - * best-effort re-queued via non-blocking {@link PollableChannel#send(Message, long)}; for other - * sources an {@link org.springframework.integration.acks.AcknowledgmentCallback} is nack'd when - * present. + * Cancellation while {@link MessageSource#receive()} is in flight (the sink is already + * {@code CANCELLED}) is handled by Reactor's discard hook on this flux. A + * {@link PollableChannel} source is best-effort re-queued via non-blocking + * {@link PollableChannel#send(Message, long)} (interceptor chain is re-entered; FIFO is not + * preserved if the queue is not empty; a full bounded channel is logged and the message is + * not nack'd). Other sources are left unacknowledged so the source's own redelivery + * (broker requeue, uncommitted offset) can recover them. The discard hook only sees drops + * at or upstream of this flux; operators a caller adds afterward + * ({@code ReactiveStreamsConsumer.setReactiveCustomizer}, a {@code flatMap} on the + * {@code ReactiveMessageHandler} path) use a different context and are not rescued here. * @param messageSource the {@link MessageSource} to adapt. * @param the expected payload type. * @return a {@link Flux} which pulls messages from the {@link MessageSource} on demand. */ @SuppressWarnings("NullAway") public static Flux> messageSourceToFlux(MessageSource messageSource) { - Consumer undeliveredConsumer = createUndeliveredConsumer(messageSource); - - return Mono.>create(monoSink -> monoSink.onRequest(request -> { - try { - monoSink.success(messageSource.receive()); - } - catch (Exception ex) { - monoSink.error(ex); - } - })) - .doOnSuccess((message) -> { - if (message != null) { - AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); - } - }) + return Mono.>create(monoSink -> + monoSink.onRequest((value) -> monoSink.success(messageSource.receive()))) .doOnError(MessagingException.class, (ex) -> { Message failedMessage = ex.getFailedMessage(); @@ -174,9 +159,10 @@ public static Flux> messageSourceToFlux(MessageSource messageS Mono.delay(ctx.getOrDefault(DELAY_WHEN_EMPTY_KEY, DEFAULT_DELAY_WHEN_EMPTY))))) .repeat() - .doOnDiscard(Message.class, message -> handleUndeliveredMessage(messageSource, message)) - .contextWrite(ctx -> ctx.put(REACTOR_ON_NEXT_DROPPED_CONTEXT_KEY, undeliveredConsumer)) - .retryWhen(Retry.indefinitely().filter(MessagingException.class::isInstance)); + .doOnDiscard(Message.class, (message) -> handleUndeliveredMessage(messageSource, message)) + .retryWhen(Retry.indefinitely().filter(MessagingException.class::isInstance)) + .doOnNext((message) -> + AckUtils.autoAck(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message))); } /** @@ -185,7 +171,8 @@ public static Flux> messageSourceToFlux(MessageSource messageS * is returned as is because it is already a {@link Publisher}; * - a {@link SubscribableChannel} is subscribed with a {@link MessageHandler} * for the {@link Sinks.Many#tryEmitNext(Object)} which is returned from this method; - * - a {@link PollableChannel} is wrapped into a {@link PollableChannelMessageSource} and reuses + * - a {@link PollableChannel} is wrapped into a {@link PollableChannelMessageSource} so + * cancelled in-flight receives can be re-queued, and reuses * {@link #messageSourceToFlux(MessageSource)}. * @param messageChannel the {@link MessageChannel} to adapt. * @param the expected payload type. @@ -208,14 +195,6 @@ else if (messageChannel instanceof PollableChannel pollableChannel) { } } - private static Consumer createUndeliveredConsumer(MessageSource messageSource) { - return value -> { - if (value instanceof Message message) { - handleUndeliveredMessage(messageSource, message); - } - }; - } - private static void handleUndeliveredMessage(MessageSource messageSource, @Nullable Message message) { if (message == null) { return; @@ -223,9 +202,6 @@ private static void handleUndeliveredMessage(MessageSource messageSource, @Nu if (messageSource instanceof PollableChannelMessageSource pollableChannelMessageSource) { pollableChannelMessageSource.returnMessage(message); } - else { - AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); - } } private static final class PollableChannelMessageSource implements MessageSource { @@ -244,9 +220,7 @@ private static final class PollableChannelMessageSource implements MessageSou void returnMessage(Message message) { if (!this.channel.send(message, 0)) { - LOGGER.warn("Failed to return undelivered message to pollable channel [" + this.channel - + "]; nacking instead"); - AckUtils.autoNack(StaticMessageHeaderAccessor.getAcknowledgmentCallback(message)); + LOGGER.warn("Failed to return undelivered message to pollable channel [" + this.channel + "]"); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java index c075de30a9d..6e9ba0ee7b1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java @@ -44,7 +44,6 @@ import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; -import static org.awaitility.Awaitility.await; /** * @author Sergei Egorov @@ -159,8 +158,9 @@ void testRetryOnMessagingExceptionOnly() { } @Test - void undeliveredMessageNackedWhenCancelledDuringBlockingReceive() throws InterruptedException { + void undeliveredMessageNotAcknowledgedWhenCancelledDuringBlockingReceive() throws InterruptedException { CountDownLatch receiveBlocked = new CountDownLatch(1); + CountDownLatch receiveCompleted = new CountDownLatch(1); AtomicBoolean releaseReceive = new AtomicBoolean(); AtomicReference ackStatus = new AtomicReference<>(); AtomicReference> delivered = new AtomicReference<>(); @@ -172,27 +172,33 @@ void undeliveredMessageNackedWhenCancelledDuringBlockingReceive() throws Interru MessageSource blockingSource = () -> { receiveBlocked.countDown(); - while (!releaseReceive.get()) { - LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); // NOSONAR busy wait + try { + while (!releaseReceive.get()) { + LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10)); // NOSONAR busy wait + } + @SuppressWarnings("unchecked") + Message message = (Message) testMessage; + return message; + } + finally { + receiveCompleted.countDown(); } - @SuppressWarnings("unchecked") - Message message = (Message) testMessage; - return message; }; Disposable subscription = IntegrationReactiveUtils.messageSourceToFlux(blockingSource) .subscribe(delivered::set); - assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); - - subscription.dispose(); - releaseReceive.set(true); - - await().atMost(Duration.ofSeconds(5)) - .until(() -> ackStatus.get() == AcknowledgmentCallback.Status.REJECT); + try { + assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); + subscription.dispose(); + } + finally { + releaseReceive.set(true); + } + assertThat(receiveCompleted.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(delivered.get()).isNull(); - assertThat(ackStatus.get()).isEqualTo(AcknowledgmentCallback.Status.REJECT); + assertThat(ackStatus.get()).isNull(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java index 75fb791cea3..b406ad1fd0f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/ReactiveStreamsConsumerTests.java @@ -293,25 +293,20 @@ public Message receive(long timeout) { verify(testSubscriber).onSubscribe(captor.capture()); captor.getValue().request(1); - assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(testChannel.getQueueSize()).isZero(); - - reactiveConsumer.stop(); - releaseReceive.set(true); + try { + assertThat(receiveBlocked.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(testChannel.getQueueSize()).isZero(); + reactiveConsumer.stop(); + } + finally { + releaseReceive.set(true); + } await().atMost(Duration.ofSeconds(5)) - .until(() -> delivered.get() != null || testChannel.getQueueSize() > 0); + .until(() -> testChannel.getQueueSize() > 0); - boolean wasDelivered = delivered.get() != null; - boolean wasRequeued = testChannel.getQueueSize() > 0; - assertThat(wasDelivered ^ wasRequeued).isTrue(); - - if (wasDelivered) { - assertThat(delivered.get()).isSameAs(testMessage); - } - else { - assertThat(testChannel.receive(0)).isSameAs(testMessage); - } + assertThat(delivered.get()).isNull(); + assertThat(testChannel.receive(0)).isSameAs(testMessage); reactiveConsumer.stop(); } From 533ebdd63b1286c4d5223893e9674b71fa71a119 Mon Sep 17 00:00:00 2001 From: Fardan An <19286898+arimu1@users.noreply.github.com> Date: Sat, 15 Aug 2026 14:55:44 +0700 Subject: [PATCH 3/3] GH-11262: Sequence no-ack test after Schedulers.single() drains receive() finally ran before monoSink.success(), so the cancel assertions could pass before discard/ack ran. Drain the single worker, then assert. Javadoc now states only messageChannelToFlux PollableChannel wrappers are re-queued. Signed-off-by: Fardan An <19286898+arimu1@users.noreply.github.com> --- .../integration/util/IntegrationReactiveUtils.java | 12 +++++++----- .../channel/IntegrationReactiveUtilsTests.java | 3 +++ 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java index 41ecd75e320..8c5ab93bfec 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/IntegrationReactiveUtils.java @@ -127,11 +127,13 @@ public static ContextView captureReactorContext() { * the {@link Mono#doOnError}. *

* Cancellation while {@link MessageSource#receive()} is in flight (the sink is already - * {@code CANCELLED}) is handled by Reactor's discard hook on this flux. A - * {@link PollableChannel} source is best-effort re-queued via non-blocking - * {@link PollableChannel#send(Message, long)} (interceptor chain is re-entered; FIFO is not - * preserved if the queue is not empty; a full bounded channel is logged and the message is - * not nack'd). Other sources are left unacknowledged so the source's own redelivery + * {@code CANCELLED}) is handled by Reactor's discard hook on this flux. Only a + * {@link PollableChannel} adapted through {@link #messageChannelToFlux(MessageChannel)} + * is best-effort re-queued via non-blocking {@link PollableChannel#send(Message, long)} + * (interceptor chain is re-entered; FIFO is not preserved if the queue is not empty; a + * full bounded channel is logged and the message is not nack'd). A direct + * {@code messageSourceToFlux(() -> channel.receive(0))} lambda is not re-queued. + * Other sources are left unacknowledged so the source's own redelivery * (broker requeue, uncommitted offset) can recover them. The discard hook only sees drops * at or upstream of this flux; operators a caller adds afterward * ({@code ReactiveStreamsConsumer.setReactiveCustomizer}, a {@code flatMap} on the diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java index 6e9ba0ee7b1..5115804dc49 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/IntegrationReactiveUtilsTests.java @@ -197,6 +197,9 @@ void undeliveredMessageNotAcknowledgedWhenCancelledDuringBlockingReceive() throw } assertThat(receiveCompleted.await(10, TimeUnit.SECONDS)).isTrue(); + CountDownLatch singleIdle = new CountDownLatch(1); + Schedulers.single().schedule(singleIdle::countDown); + assertThat(singleIdle.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(delivered.get()).isNull(); assertThat(ackStatus.get()).isNull(); }