This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 438dd1b1f922 CAMEL-25503: camel-reactive-streams - Make the consumer
Suspendable
438dd1b1f922 is described below
commit 438dd1b1f922ed1c81ca2da2c3d11ac29388af9e
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 12:20:07 2026 +0530
CAMEL-25503: camel-reactive-streams - Make the consumer Suspendable
Suspending a route with a reactive-streams consumer stopped the consumer,
which waited for its queued items to drain, and a route policy that
suspended it from one of its own exchanges made those items fail. The
consumer is now Suspendable: a suspend requests no more items from the
publisher and keeps the items already received queued, and routes them in
order when the consumer is resumed or stopped. A request sent with
toStream to a suspended route now waits for the resume instead of failing.
The consumer is also ShutdownAware, so a graceful shutdown (stopping the
route, all routes or the CamelContext) still routes the queued items, and
waits for them within the shutdown timeout, before it stops the downstream
direct or seda routes they go to. No more items are requested while
draining.
Closes #27636
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/docs/reactive-streams-component.adoc | 16 ++
.../src/main/docs/reactive-streams-component.adoc | 16 ++
.../streams/ReactiveStreamsCamelSubscriber.java | 4 +-
.../reactive/streams/ReactiveStreamsConsumer.java | 192 +++++++++++++++---
.../streams/ConsumerGracefulShutdownTest.java | 135 +++++++++++++
.../reactive/streams/ConsumerStopTest.java | 4 +-
.../reactive/streams/ConsumerSuspendTest.java | 216 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 11 ++
8 files changed, 570 insertions(+), 24 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/reactive-streams-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/reactive-streams-component.adoc
index 8ce07c72b29d..4d78cf3dea88 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/reactive-streams-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/reactive-streams-component.adoc
@@ -591,3 +591,19 @@ is always lower than `10`. Messages are processed by a
single thread in the Came
The number of concurrent consumers (threads) can also be set as endpoint
option (`concurrentConsumers`).
When using 1 consumer (the default), the order of items in the source stream
is maintained.
When this value is increased, items will be processed concurrently by multiple
threads (so not preserving the order).
+
+==== Suspending the consumer
+
+The consumer can be suspended: by suspending its route (route controller,
JMX), by a route policy such as
+`ThrottlingInflightRoutePolicy`, and during a graceful shutdown. A suspended
consumer requests no more items from the
+publisher and does not route the items it has already received (the items
queued for its threads, and the items the
+publisher still sends for the demand requested before the suspend). The
exchanges being routed complete normally, and
+the suspend does not wait for the queued items: they are routed in order when
the consumer is resumed, or when it is
+stopped.
+
+A graceful shutdown (stopping the route, all the routes or the CamelContext)
also suspends the consumer first, but then
+routes its queued items, and waits for them up to the shutdown timeout, before
it stops the routes the items go to, such
+as a `direct` or `seda` route. No more items are requested from the publisher.
+
+When backpressure is disabled (`maxInflightExchanges` not positive), a
suspended consumer cannot limit what the
+publisher sends: the items it receives while suspended are kept in memory
until it is resumed or stopped.
diff --git
a/components/camel-reactive-streams/src/main/docs/reactive-streams-component.adoc
b/components/camel-reactive-streams/src/main/docs/reactive-streams-component.adoc
index 8ce07c72b29d..4d78cf3dea88 100644
---
a/components/camel-reactive-streams/src/main/docs/reactive-streams-component.adoc
+++
b/components/camel-reactive-streams/src/main/docs/reactive-streams-component.adoc
@@ -591,3 +591,19 @@ is always lower than `10`. Messages are processed by a
single thread in the Came
The number of concurrent consumers (threads) can also be set as endpoint
option (`concurrentConsumers`).
When using 1 consumer (the default), the order of items in the source stream
is maintained.
When this value is increased, items will be processed concurrently by multiple
threads (so not preserving the order).
+
+==== Suspending the consumer
+
+The consumer can be suspended: by suspending its route (route controller,
JMX), by a route policy such as
+`ThrottlingInflightRoutePolicy`, and during a graceful shutdown. A suspended
consumer requests no more items from the
+publisher and does not route the items it has already received (the items
queued for its threads, and the items the
+publisher still sends for the demand requested before the suspend). The
exchanges being routed complete normally, and
+the suspend does not wait for the queued items: they are routed in order when
the consumer is resumed, or when it is
+stopped.
+
+A graceful shutdown (stopping the route, all the routes or the CamelContext)
also suspends the consumer first, but then
+routes its queued items, and waits for them up to the shutdown timeout, before
it stops the routes the items go to, such
+as a `direct` or `seda` route. No more items are requested from the publisher.
+
+When backpressure is disabled (`maxInflightExchanges` not positive), a
suspended consumer cannot limit what the
+publisher sends: the items it receives while suspended are kept in memory
until it is resumed or stopped.
diff --git
a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
index 444fb42ef675..c2ae81ea2afb 100644
---
a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
+++
b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsCamelSubscriber.java
@@ -154,7 +154,9 @@ public class ReactiveStreamsCamelSubscriber implements
Subscriber<Exchange>, Clo
Subscription subs = null;
lock.lock();
try {
- if (consumer != null && this.subscription != null) {
+ // a suspended consumer requests nothing until it is resumed, nor
does a consumer drained by a shutdown
+ if (consumer != null && this.subscription != null &&
!consumer.isSuspendingOrSuspended()
+ && !consumer.isDraining()) {
Integer consMax =
consumer.getEndpoint().getMaxInflightExchanges();
long max = (consMax != null && consMax > 0) ?
consMax.longValue() : UNBOUNDED_REQUESTS;
if (requested < UNBOUNDED_REQUESTS) {
diff --git
a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
index f91c68e2ed09..86e9f052ba9f 100644
---
a/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
+++
b/components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java
@@ -16,13 +16,19 @@
*/
package org.apache.camel.component.reactive.streams;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.Suspendable;
import
org.apache.camel.component.reactive.streams.api.CamelReactiveStreamsService;
+import org.apache.camel.spi.ShutdownAware;
import org.apache.camel.support.DefaultConsumer;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -30,8 +36,17 @@ import org.slf4j.LoggerFactory;
/**
* The Camel reactive-streams consumer.
+ * <p/>
+ * The items received from the stream are queued and routed by the threads of
the consumer. A suspended consumer
+ * requests no more items from the stream and routes none of the queued ones
(nor the items the publisher still sends
+ * for the demand requested before the suspend): they stay queued until the
consumer is resumed or stopped. The
+ * exchanges being routed when the consumer is suspended complete normally,
and the suspend does not wait for them.
+ * <p/>
+ * A graceful shutdown also suspends the consumer first, but then routes the
queued items while the other routes are
+ * still running (see {@link #getPendingExchangesSize(boolean)}), so that they
can reach a downstream route such as a
+ * direct or seda one, and the shutdown strategy waits for them within its
timeout.
*/
-public class ReactiveStreamsConsumer extends DefaultConsumer {
+public class ReactiveStreamsConsumer extends DefaultConsumer implements
Suspendable, ShutdownAware {
private static final Logger LOG =
LoggerFactory.getLogger(ReactiveStreamsConsumer.class);
@@ -43,7 +58,22 @@ public class ReactiveStreamsConsumer extends DefaultConsumer
{
private final ReactiveStreamsEndpoint endpoint;
private final CamelReactiveStreamsService service;
+ /**
+ * The items received from the stream that are not routed yet, in the
order they were received. Each one has a task
+ * in the thread pool that routes the oldest queued item, unless the
consumer is suspended.
+ */
+ private final Queue<QueuedItem> queued = new ConcurrentLinkedQueue<>();
+ /**
+ * The number of items taken from {@link #queued} whose exchange is not
done yet.
+ */
+ private final AtomicInteger routing = new AtomicInteger();
+ /**
+ * Whether a graceful shutdown is draining the queued items: they are
routed even though the consumer is suspended,
+ * while no more items are requested from the stream.
+ */
+ private volatile boolean draining;
private ExecutorService executor;
+ private ReactiveStreamsCamelSubscriber subscriber;
public ReactiveStreamsConsumer(ReactiveStreamsEndpoint endpoint, Processor
processor, CamelReactiveStreamsService service) {
super(endpoint, processor);
@@ -54,6 +84,7 @@ public class ReactiveStreamsConsumer extends DefaultConsumer {
@Override
protected void doStart() throws Exception {
super.doStart();
+ draining = false;
int poolSize = endpoint.getConcurrentConsumers();
if (executor == null) {
@@ -61,17 +92,48 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
getEndpoint().getEndpointUri(), poolSize);
}
- this.service.attachCamelConsumer(endpoint.getStream(), this);
+ // the items left queued by a stop that timed out are routed now
+ scheduleQueuedItems();
+
+ this.subscriber =
this.service.attachCamelConsumer(endpoint.getStream(), this);
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ // nothing to wait for: while the consumer is suspended, its
subscriber requests no more items from the stream
+ // and the queued items are not routed (see routeQueuedItem); the
exchanges being routed complete normally
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ draining = false;
+ if (executor == null) {
+ // suspended while it was not started (before its start or after a
stop). Call doStart() directly, not
+ // start(): resume() has already set the status to STARTING, so
start() would return without starting
+ // the consumer. resume() sets the status to STARTED once this
returns.
+ doStart();
+ return;
+ }
+
+ // route the items queued while suspended, then request more items
from the stream
+ scheduleQueuedItems();
+ ReactiveStreamsCamelSubscriber current = this.subscriber;
+ if (current != null) {
+ current.refill();
+ }
}
@Override
protected void doStop() throws Exception {
this.service.detachCamelConsumer(endpoint.getStream());
+ this.subscriber = null;
if (executor != null) {
// the queued exchanges were already taken from the stream (they
cannot be requested again),
// so let them complete before the processor is stopped; dropping
them would also leave
// them counted as inflight by the subscriber, which would then
request less or nothing
+ // (the tasks of the items queued while the consumer was suspended
did not route them)
+ scheduleQueuedItems();
if (ROUTING.get() == this) {
// stopped by an exchange of this consumer (for example a
route policy): the pool cannot be
// awaited from one of its own threads, the queued exchanges
run after this one
@@ -81,10 +143,62 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
}
executor = null;
}
+ draining = false;
super.doStop();
}
+ @Override
+ public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
+ // the shutdown strategy suspends this consumer in its first pass, and
waits for the queued items to be routed
+ // (see getPendingExchangesSize) while the routes they go to are still
running
+ return false;
+ }
+
+ @Override
+ public int getPendingExchangesSize() {
+ return executor != null ? queued.size() + routing.get() : 0;
+ }
+
+ /**
+ * Returns the number of items the shutdown strategy must wait for.
+ * <p/>
+ * When the route is being shutdown, this starts to route the queued
items, even though the consumer is suspended,
+ * and returns the number of items queued or being routed: the strategy
waits for them (up to its timeout) before it
+ * stops the deferred consumers, such as the direct or seda consumers of
the routes the items go to. No more items
+ * are requested from the stream. When the route is only being suspended,
the queued items are kept until the
+ * consumer is resumed, so there is nothing to wait for.
+ */
+ @Override
+ public int getPendingExchangesSize(boolean suspendOnly) {
+ if (suspendOnly) {
+ return 0;
+ }
+ startDraining();
+ return getPendingExchangesSize();
+ }
+
+ @Override
+ public void prepareShutdown(boolean suspendOnly, boolean forced) {
+ if (!suspendOnly && !forced) {
+ // the queued items are routed before the consumer is stopped (a
no-op when the shutdown strategy already
+ // started it while it waited for the pending exchanges)
+ startDraining();
+ }
+ }
+
+ boolean isDraining() {
+ return draining;
+ }
+
+ private void startDraining() {
+ if (!draining && executor != null) {
+ draining = true;
+ // the tasks of the items queued while the consumer was suspended
did not route them
+ scheduleQueuedItems();
+ }
+ }
+
public boolean process(Exchange exchange, AsyncCallback callback) {
exchange.getIn().setHeader(ReactiveStreamsConstants.REACTIVE_STREAMS_EVENT_TYPE,
"onNext");
return doSend(exchange, callback);
@@ -115,26 +229,8 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
ExecutorService executorService = this.executor;
if (executorService != null && this.isRunAllowed()) {
- executorService.execute(() -> {
- ROUTING.set(this);
- try {
- this.getAsyncProcessor().process(exchange, doneSync -> {
- Exception cause = exchange.getException();
- if (cause instanceof RejectedExecutionException &&
!isRunAllowed()) {
- // an item queued when the consumer was stopped
from one of its own exchanges: one line
- // per item, as there can be up to
maxInflightExchanges of them
- LOG.warn("Item {} of stream {} not routed as the
consumer is stopped",
- exchange.getExchangeId(),
endpoint.getStream());
- } else if (cause != null) {
- getExceptionHandler().handleException("Error
processing exchange", exchange, cause);
- }
-
- callback.done(doneSync);
- });
- } finally {
- ROUTING.remove();
- }
- });
+ queued.add(new QueuedItem(exchange, callback));
+ executorService.execute(this::routeQueuedItem);
return false;
} else {
@@ -144,9 +240,61 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
}
}
+ /**
+ * Adds a task per queued item to the thread pool: the tasks of the items
queued while the consumer was suspended
+ * did not route them. A task that finds no item left does nothing.
+ */
+ private void scheduleQueuedItems() {
+ ExecutorService executorService = this.executor;
+ if (executorService == null) {
+ return;
+ }
+ for (int i = queued.size(); i > 0; i--) {
+ executorService.execute(this::routeQueuedItem);
+ }
+ }
+
+ private void routeQueuedItem() {
+ if (isSuspendingOrSuspended() && !draining) {
+ // the item stays queued until the consumer is resumed, stopped or
drained by a graceful shutdown
+ return;
+ }
+ // counted before it is taken from the queue, so that the shutdown
strategy always sees it as pending
+ routing.incrementAndGet();
+ QueuedItem item = queued.poll();
+ if (item == null) {
+ routing.decrementAndGet();
+ return;
+ }
+
+ Exchange exchange = item.exchange();
+ ROUTING.set(this);
+ try {
+ this.getAsyncProcessor().process(exchange, doneSync -> {
+ Exception cause = exchange.getException();
+ if (cause instanceof RejectedExecutionException &&
!isRunAllowed()) {
+ // an item queued when the consumer was stopped from one
of its own exchanges: one line
+ // per item, as there can be up to maxInflightExchanges of
them
+ LOG.warn("Item {} of stream {} not routed as the consumer
is stopped",
+ exchange.getExchangeId(), endpoint.getStream());
+ } else if (cause != null) {
+ getExceptionHandler().handleException("Error processing
exchange", exchange, cause);
+ }
+
+ routing.decrementAndGet();
+ item.callback().done(doneSync);
+ });
+ } finally {
+ ROUTING.remove();
+ }
+ }
+
@Override
public ReactiveStreamsEndpoint getEndpoint() {
return endpoint;
}
+ private record QueuedItem(Exchange exchange, AsyncCallback callback) {
+ }
+
}
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerGracefulShutdownTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerGracefulShutdownTest.java
new file mode 100644
index 000000000000..2e0a7c03e8a7
--- /dev/null
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerGracefulShutdownTest.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.reactive.streams;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.support.service.ServiceSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A graceful shutdown of several routes suspends the consumer first, and must
route the items it has queued while the
+ * downstream direct consumer is still running: the routes are shutdown in
reverse startup order, so the direct consumer
+ * is stopped before the reactive-streams one.
+ */
+class ConsumerGracefulShutdownTest extends BaseReactiveTest {
+
+ private final CountDownLatch gateReached = new CountDownLatch(1);
+ private final CountDownLatch gate = new CountDownLatch(1);
+ private final CountDownLatch blockSecond = new CountDownLatch(1);
+ private final List<Integer> received = new CopyOnWriteArrayList<>();
+ private volatile boolean blockSecondItem;
+
+ @AfterEach
+ void openGates() {
+ gate.countDown();
+ blockSecond.countDown();
+ }
+
+ @Test
+ void testContextStopRoutesTheQueuedItemsToTheDirectRoute() throws
Exception {
+ startWithQueuedItems();
+
+ stopContextWhileTheFirstItemWaits();
+
+ assertEquals(IntStream.rangeClosed(1, 10).boxed().toList(), received,
+ "The items queued in the consumer must reach the direct route
before it is stopped");
+ }
+
+ @Test
+ void testShutdownTimeoutAppliesToTheQueuedItems() throws Exception {
+ // the second item blocks in the direct route: the shutdown strategy
waits for the queued items up to its
+ // timeout, then forces the routes to stop
+ context.getShutdownStrategy().setTimeout(2);
+ context.getExecutorServiceManager().setShutdownAwaitTermination(1000);
+ blockSecondItem = true;
+ startWithQueuedItems();
+
+ stopContextWhileTheFirstItemWaits();
+
+ assertTrue(context.getShutdownStrategy().hasTimeoutOccurred(),
+ "The shutdown strategy must wait for the queued items until
its timeout");
+ assertTrue(received.contains(2), "The queued items must be routed
during the graceful shutdown");
+ }
+
+ private void startWithQueuedItems() {
+ Flux.range(1,
1000).subscribe(CamelReactiveStreams.get(context).streamSubscriber("in",
Integer.class));
+
+ // the publisher sent the 10 requested numbers: the first one waits in
the direct route, the others are queued
+ assertTrue(awaitGate(), "The first item must reach the direct route");
+ ReactiveStreamsCamelSubscriber subscriber
+ = (ReactiveStreamsCamelSubscriber)
CamelReactiveStreams.get(context).streamSubscriber("in");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
subscriber.getInflightCount() == 10);
+ }
+
+ private boolean awaitGate() {
+ try {
+ return gateReached.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ private void stopContextWhileTheFirstItemWaits() throws Exception {
+ CompletableFuture<Void> stop =
CompletableFuture.runAsync(context::stop);
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute("rs").getConsumer();
+ // the graceful shutdown suspends the consumer first
+ await().atMost(10, TimeUnit.SECONDS).until(consumer::isSuspended);
+ gate.countDown();
+ stop.get(30, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // defined (and started) before the direct route, so it is
shutdown after it
+
from("reactive-streams:in?maxInflightExchanges=10").routeId("rs")
+ .to("direct:sub");
+
+ from("direct:sub").routeId("sub")
+ .process(e -> {
+ int item = e.getMessage().getBody(Integer.class);
+ if (item == 1) {
+ gateReached.countDown();
+ gate.await();
+ } else if (item == 2 && blockSecondItem) {
+ received.add(item);
+ blockSecond.await();
+ return;
+ }
+ received.add(item);
+ });
+ }
+ };
+ }
+}
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
index ce42028f8e13..b57277df57d6 100644
---
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
@@ -104,7 +104,9 @@ class ConsumerStopTest extends BaseReactiveTest {
}
});
ServiceSupport consumer = (ServiceSupport)
context.getRoute(routeId).getConsumer();
- await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.isStopping()
|| consumer.isStopped());
+ // the graceful stop suspends the consumer first, and stops it once
the exchange in flight is done
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> consumer.isSuspended() || consumer.isStopping()
|| consumer.isStopped());
gate.countDown();
stop.get(20, TimeUnit.SECONDS);
}
diff --git
a/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerSuspendTest.java
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerSuspendTest.java
new file mode 100644
index 000000000000..28b2c5a31fc1
--- /dev/null
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerSuspendTest.java
@@ -0,0 +1,216 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.reactive.streams;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.reactive.streams.api.CamelReactiveStreams;
+import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.support.service.ServiceSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A suspended consumer (suspended route, route policies, graceful shutdown)
routes no new items and does not wait for
+ * the items it already took from the stream; they are routed when it is
resumed or stopped.
+ */
+class ConsumerSuspendTest extends BaseReactiveTest {
+
+ private final CountDownLatch gateReached = new CountDownLatch(1);
+ private final CountDownLatch gate = new CountDownLatch(1);
+ private final CountDownLatch policySuspended = new CountDownLatch(1);
+
+ private final RoutePolicySupport suspendingPolicy = new
RoutePolicySupport() {
+ @Override
+ public void onExchangeDone(Route route, Exchange exchange) {
+ // on the thread of the consumer, as ThrottlingInflightRoutePolicy
does
+ if (exchange.getMessage().getBody(Integer.class) == 3) {
+ try {
+ suspendOrStopConsumer(route.getConsumer());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ policySuspended.countDown();
+ }
+ }
+ };
+
+ @AfterEach
+ void openGate() {
+ // a suspend that waits for the exchange behind the gate must not
block the context stop
+ gate.countDown();
+ }
+
+ @Test
+ void testSuspendDoesNotWaitForTheQueuedExchanges() throws Exception {
+ // a suspend that drained the queued exchanges would wait for the gate
up to this timeout
+ context.getExecutorServiceManager().setShutdownAwaitTermination(60000);
+ Flux.range(1,
1000).subscribe(CamelReactiveStreams.get(context).streamSubscriber("queued",
Integer.class));
+ // the publisher sent the 10 requested numbers: the first one waits in
the route, the others are queued
+ assertTrue(gateReached.await(10, TimeUnit.SECONDS));
+ ReactiveStreamsCamelSubscriber subscriber
+ = (ReactiveStreamsCamelSubscriber)
CamelReactiveStreams.get(context).streamSubscriber("queued");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
subscriber.getInflightCount() == 10);
+
+ Consumer consumer = context.getRoute("queued").getConsumer();
+ CompletableFuture.runAsync(() ->
ServiceHelper.suspendService(consumer)).get(10, TimeUnit.SECONDS);
+ assertTrue(((ServiceSupport) consumer).isSuspended(), "The consumer
must be suspended");
+
+ // the exchange in flight completes, the queued ones are not routed
while the consumer is suspended
+ MockEndpoint mock = getMockEndpoint("mock:queued");
+ mock.expectedBodiesReceived(1);
+ mock.setAssertPeriod(500);
+ gate.countDown();
+ mock.assertIsSatisfied();
+ // and no more items are requested from the stream (the refill
watermark would request one per exchange done)
+ assertEquals(9, subscriber.getInflightCount(), "The queued items must
be held");
+ assertEquals(0, subscriber.getRequested(), "A suspended consumer must
not request items");
+
+ // a stop routes the items the suspended consumer took from the stream
+ mock.reset();
+ mock.expectedBodiesReceived(IntStream.rangeClosed(2,
10).boxed().toList());
+ context.getRouteController().stopRoute("queued");
+ mock.assertIsSatisfied();
+ assertEquals(0, subscriber.getInflightCount());
+ }
+
+ @Test
+ void testSuspendAndResumeRoute() throws Exception {
+ Sinks.Many<Integer> sink =
Sinks.many().unicast().onBackpressureBuffer();
+
sink.asFlux().subscribe(CamelReactiveStreams.get(context).streamSubscriber("controlled",
Integer.class));
+ MockEndpoint mock = getMockEndpoint("mock:controlled");
+ mock.expectedBodiesReceived(1, 2, 3, 4, 5);
+ IntStream.rangeClosed(1, 5).forEach(sink::tryEmitNext);
+ mock.assertIsSatisfied();
+
+ context.getRouteController().suspendRoute("controlled");
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus("controlled"));
+ assertTrue(((ServiceSupport)
context.getRoute("controlled").getConsumer()).isSuspended(),
+ "The consumer must be suspended");
+
+ // the items sent for the outstanding demand are held, the others stay
with the publisher
+ mock.reset();
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(500);
+ IntStream.rangeClosed(6, 30).forEach(sink::tryEmitNext);
+ mock.assertIsSatisfied();
+
+ mock.reset();
+ mock.setAssertPeriod(0);
+ mock.expectedBodiesReceived(IntStream.rangeClosed(6,
30).boxed().toList());
+ context.getRouteController().resumeRoute("controlled");
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("controlled"));
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testRequestToASuspendedRouteWaitsForTheResume() throws Exception {
+ context.getRouteController().suspendRoute("controlled");
+ MockEndpoint mock = getMockEndpoint("mock:controlled");
+ mock.expectedMessageCount(0);
+ mock.setAssertPeriod(500);
+
+ CompletableFuture<Exchange> reply =
Mono.from(CamelReactiveStreams.get(context).toStream("controlled",
7)).toFuture();
+ mock.assertIsSatisfied();
+ assertFalse(reply.isDone(), "A suspended route must not process the
request");
+
+ mock.reset();
+ mock.setAssertPeriod(0);
+ mock.expectedBodiesReceived(7);
+ context.getRouteController().resumeRoute("controlled");
+ mock.assertIsSatisfied();
+ assertEquals(7, reply.get(10,
TimeUnit.SECONDS).getMessage().getBody(Integer.class));
+ }
+
+ @Test
+ void testRoutePolicySuspendsAndResumesTheConsumer() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:policy");
+ mock.expectedBodiesReceived(1, 2, 3);
+ mock.setAssertPeriod(500);
+ Flux.range(1,
50).subscribe(CamelReactiveStreams.get(context).streamSubscriber("policy",
Integer.class));
+ assertTrue(policySuspended.await(10, TimeUnit.SECONDS), "The route
policy must suspend the consumer");
+ Consumer consumer = context.getRoute("policy").getConsumer();
+ assertTrue(((ServiceSupport) consumer).isSuspended(), "The consumer
must be suspended");
+ mock.assertIsSatisfied();
+
+ // the items taken from the stream before the suspend are routed after
the resume, in order
+ mock.reset();
+ mock.setAssertPeriod(0);
+ mock.expectedBodiesReceived(IntStream.rangeClosed(4,
50).boxed().toList());
+ suspendingPolicy.resumeOrStartConsumer(consumer);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void testResumeStartsAConsumerThatWasNotStarted() throws Exception {
+ Consumer consumer = context.getRoute("notStarted").getConsumer();
+ Flux.range(1,
5).subscribe(CamelReactiveStreams.get(context).streamSubscriber("notStarted",
Integer.class));
+ MockEndpoint mock = getMockEndpoint("mock:notStarted");
+ mock.expectedBodiesReceived(1, 2, 3, 4, 5);
+
+ ServiceHelper.suspendService(consumer);
+ ServiceHelper.resumeService(consumer);
+
+ mock.assertIsSatisfied();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> ((ServiceSupport)
consumer).isStarted());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("reactive-streams:queued?maxInflightExchanges=10&exchangesRefillLowWatermark=1").routeId("queued")
+ .process(e -> {
+ if (e.getMessage().getBody(Integer.class) == 1) {
+ gateReached.countDown();
+ gate.await();
+ }
+ })
+ .to("mock:queued");
+
+
from("reactive-streams:controlled?maxInflightExchanges=10").routeId("controlled")
+ .to("mock:controlled");
+
+
from("reactive-streams:policy?maxInflightExchanges=10").routeId("policy")
+ .routePolicy(suspendingPolicy)
+ .to("mock:policy");
+
+
from("reactive-streams:notStarted").routeId("notStarted").autoStartup(false)
+ .to("mock:notStarted");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index ff5a398357c5..bd76cca82786 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -1119,6 +1119,17 @@ these items were dropped, and after a restart the
consumer could stop requesting
stops itself from one of its own exchanges now waits for that timeout; stop it
asynchronously instead (for example with
the Control Bus `async=true` option).
+=== camel-reactive-streams - suspending a consumer
+
+The `reactive-streams` consumer is now suspendable: suspending its route
(route controller, JMX) and a route policy
+such as `ThrottlingInflightRoutePolicy` suspend it instead of stopping it, and
a graceful shutdown suspends it before
+stopping it. It then requests no more items and holds the items it has already
received, without waiting for them, and
+routes them in order when it is resumed (or stopped). Before, the consumer was
stopped (the route status was `Stopped` instead of `Suspended`), which
+waited for those items, and a route policy that suspended it from one of its
exchanges made them fail. A request sent
+with `toStream` to a suspended route now waits for the resume instead of
failing.
+A graceful shutdown still routes the queued items before it stops the other
routes, such as a `direct` route they go
+to.
+
=== camel-vertx - request/reply over the event bus
When a `vertx` consumer receives a message that expects a reply and the route
fails, the sender now gets a failure
