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 410f1ef98e5d CAMEL-25351: camel-reactive-streams - a stopped consumer
must process the exchanges it already took from the stream (#27431)
410f1ef98e5d is described below
commit 410f1ef98e5dca12b70985eb4ff0c10cffb7ae76
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 6 18:47:31 2026 +0530
CAMEL-25351: camel-reactive-streams - a stopped consumer must process the
exchanges it already took from the stream (#27431)
* CAMEL-25351: camel-reactive-streams - a stopped consumer must process the
exchanges it already took from the stream
The reactive streams consumer hands each item it receives from the stream
to its own thread pool (concurrentConsumers threads), and its subscriber
counts it as inflight until the exchange is done; the subscriber requests
more items only while fewer than maxInflightExchanges are requested or
inflight.
ReactiveStreamsConsumer.doStop() stopped the route processor and then shut
the thread pool down with shutdownNow(). The consumer is not Suspendable, so
the graceful shutdown stops it right away (route stop, CamelContext stop,
route policies): the items queued in the pool were dropped without any log
(they cannot be requested from the publisher again), the running one was
interrupted, and the subscriber kept the dropped ones as inflight for good.
When the route is started again the subscriber requests less, or nothing at
all once the leaked count is above the refill watermark (more than 32 with
the default maxInflightExchanges=128): the route never consumes again.
doStop() now detaches the consumer from the stream first (new items are
discarded with a WARN, as before), then shuts the pool down gracefully, so
the queued exchanges are routed before the route processor is stopped by
DefaultConsumer.doStop().
A route policy that stops the consumer when an exchange is done (for example
ThrottlingInflightRoutePolicy or ThrottlingExceptionRoutePolicy) runs on a
thread of the consumer's pool. Waiting for the pool to terminate from that
thread waited for itself until shutdownAwaitTermination (10 s by default),
then interrupted the thread and dropped the queued exchanges.
doStop() now only shuts the pool down when it is called while routing an
exchange of this consumer: the queued exchanges run after the current one
and
complete (failed with a RejectedExecutionException once the route processor
is
stopped, or routed if the consumer is started again first), so they are not
left inflight in the subscriber.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25351: camel-reactive-streams - Shorter upgrade note, log a stopped
item in one line, drop the control test
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../reactive/streams/ReactiveStreamsConsumer.java | 46 +++++--
.../reactive/streams/ConsumerStopTest.java | 152 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 8 ++
3 files changed, 198 insertions(+), 8 deletions(-)
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 989b9012fca8..f91c68e2ed09 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
@@ -17,6 +17,7 @@
package org.apache.camel.component.reactive.streams;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -34,6 +35,12 @@ public class ReactiveStreamsConsumer extends DefaultConsumer
{
private static final Logger LOG =
LoggerFactory.getLogger(ReactiveStreamsConsumer.class);
+ /**
+ * The consumer whose thread pool runs the exchange that the current
thread is routing. It is only set while the
+ * pool thread routes the exchange synchronously.
+ */
+ private static final ThreadLocal<ReactiveStreamsConsumer> ROUTING = new
ThreadLocal<>();
+
private final ReactiveStreamsEndpoint endpoint;
private final CamelReactiveStreamsService service;
private ExecutorService executor;
@@ -59,13 +66,23 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
@Override
protected void doStop() throws Exception {
- super.doStop();
this.service.detachCamelConsumer(endpoint.getStream());
if (executor != null) {
-
endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(executor);
+ // 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
+ 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
+
endpoint.getCamelContext().getExecutorServiceManager().shutdown(executor);
+ } else {
+
endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(executor);
+ }
executor = null;
}
+
+ super.doStop();
}
public boolean process(Exchange exchange, AsyncCallback callback) {
@@ -98,13 +115,26 @@ public class ReactiveStreamsConsumer extends
DefaultConsumer {
ExecutorService executorService = this.executor;
if (executorService != null && this.isRunAllowed()) {
- executorService.execute(() ->
this.getAsyncProcessor().process(exchange, doneSync -> {
- if (exchange.getException() != null) {
- getExceptionHandler().handleException("Error processing
exchange", exchange, exchange.getException());
+ 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();
}
-
- callback.done(doneSync);
- }));
+ });
return false;
} else {
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
new file mode 100644
index 000000000000..ce42028f8e13
--- /dev/null
+++
b/components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.Exchange;
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+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.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;
+
+class ConsumerStopTest extends BaseReactiveTest {
+
+ private final CountDownLatch gateReached = new CountDownLatch(1);
+ private final CountDownLatch gate = new CountDownLatch(1);
+ private final CountDownLatch policyStopped = new CountDownLatch(1);
+ private final List<Integer> received = new CopyOnWriteArrayList<>();
+
+ @Test
+ void testStopProcessesTheExchangesTakenFromTheStream() throws Exception {
+ startWithQueuedExchanges("queued", 1);
+
+ stopWhileTheFirstExchangeWaits("queued");
+
+ assertEquals(IntStream.rangeClosed(1, 10).boxed().toList(), received,
+ "The exchanges taken from the stream before the stop must be
processed");
+ }
+
+ @Test
+ void testConsumesAgainAfterRestart() throws Exception {
+ startWithQueuedExchanges("queued", 1);
+ stopWhileTheFirstExchangeWaits("queued");
+
+ context.getRouteController().startRoute("queued");
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertTrue(received.contains(11), "The route must
consume again after its restart"));
+ }
+
+ @Test
+ void testStopFromTheRouteDoesNotWaitForItself() throws Exception {
+ // a stop that waited for the pool from one of its threads would wait
for this timeout
+ context.getExecutorServiceManager().setShutdownAwaitTermination(60000);
+ startWithQueuedExchanges("policy", 1);
+
+ gate.countDown();
+
+ assertTrue(policyStopped.await(20, TimeUnit.SECONDS), "The route
policy must be able to stop the consumer");
+ // the exchanges queued behind the one that stopped the consumer still
complete
+ ReactiveStreamsCamelSubscriber subscriber
+ = (ReactiveStreamsCamelSubscriber)
CamelReactiveStreams.get(context).streamSubscriber("policy");
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
subscriber.getInflightCount() == 0);
+
+ ServiceHelper.startService(context.getRoute("policy").getConsumer());
+
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertTrue(received.contains(11), "The consumer must
consume again after its restart"));
+ }
+
+ private void startWithQueuedExchanges(String stream, int waiting) throws
Exception {
+ Flux.range(1,
1000).subscribe(CamelReactiveStreams.get(context).streamSubscriber(stream,
Integer.class));
+
+ // the publisher sent the 10 requested numbers: one waits in the
route, the others are queued or done
+ assertTrue(gateReached.await(10, TimeUnit.SECONDS));
+ ReactiveStreamsCamelSubscriber subscriber
+ = (ReactiveStreamsCamelSubscriber)
CamelReactiveStreams.get(context).streamSubscriber(stream);
+ await().atMost(10, TimeUnit.SECONDS).until(() ->
subscriber.getInflightCount() == 11 - waiting);
+ }
+
+ private void stopWhileTheFirstExchangeWaits(String routeId) throws
Exception {
+ CompletableFuture<Void> stop = CompletableFuture.runAsync(() -> {
+ try {
+ context.getRouteController().stopRoute(routeId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ ServiceSupport consumer = (ServiceSupport)
context.getRoute(routeId).getConsumer();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> consumer.isStopping()
|| consumer.isStopped());
+ gate.countDown();
+ stop.get(20, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("reactive-streams:queued?maxInflightExchanges=10").routeId("queued")
+ .process(e -> {
+ if (e.getMessage().getBody(Integer.class) == 1) {
+ gateReached.countDown();
+ gate.await();
+ }
+
received.add(e.getMessage().getBody(Integer.class));
+ });
+
+ // a route policy that stops the consumer when the first
exchange is done, on the thread of the consumer
+
from("reactive-streams:policy?maxInflightExchanges=10").routeId("policy")
+ .routePolicy(new RoutePolicySupport() {
+ @Override
+ public void onExchangeDone(Route route, Exchange
exchange) {
+ if
(exchange.getMessage().getBody(Integer.class) == 1) {
+ try {
+ stopConsumer(route.getConsumer());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ policyStopped.countDown();
+ }
+ }
+ })
+ .process(e -> {
+ if (e.getMessage().getBody(Integer.class) == 1) {
+ gateReached.countDown();
+ gate.await();
+ }
+
received.add(e.getMessage().getBody(Integer.class));
+ });
+
+ }
+ };
+ }
+}
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 e553f7c299bc..bfb4845a5b4d 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
@@ -957,6 +957,14 @@ not answer within the timeout of `receive(timeout)` or
`receiveNoWait()`, as oth
periodically and should go on while the PLC is unreachable can handle the
exception, for example with `onException`
or `doTry`/`doCatch`.
+=== camel-reactive-streams - stopping a consumer
+
+Stopping a `reactive-streams` consumer now routes the items it already
received from the stream before the route
+stops, waiting up to the `shutdownAwaitTermination` of the
`ExecutorServiceManager` (10 seconds by default). Before,
+these items were dropped, and after a restart the consumer could stop
requesting items. A route that synchronously
+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-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
