allthingssecurity opened a new pull request, #27431: URL: https://github.com/apache/camel/pull/27431
# Description [CAMEL-25351](https://issues.apache.org/jira/browse/CAMEL-25351) `ReactiveStreamsConsumer` queues each item it receives from the stream on its own thread pool, and `ReactiveStreamsCamelSubscriber` counts it as inflight until the exchange is done; it requests more only while fewer than `maxInflightExchanges` are requested or inflight, in batches above the refill watermark. `doStop()` stopped the route processor (`super.doStop()`), then called `shutdownNow` on the pool. The consumer is not `Suspendable`, so a graceful shutdown (route stop, CamelContext stop, route policies) stops it right away: the queued items were dropped without a 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. After a restart it requests fewer items, or none at all once more than 32 were dropped with the defaults (`128 - dropped < 96`): the route never consumes again. camel-reactor and camel-rxjava use the same consumer and subscriber. This change detaches the consumer first (later items are discarded with a WARN, as before), shuts the pool down with `shutdownGraceful` so that the queued exchanges are routed (up to `shutdownAwaitTermination`, 10 s by default), and only then calls `super.doStop()`. Draining after `super.doStop()` is not enough: the stopped error handler rejects the drained exchanges with a `RejectedExecutionException`. A stop called on a thread of the consumer's own pool, such as a route policy that stops the consumer when an exchange is done (`RoutePolicySupport.stopConsumer`, used by `ThrottlingInflightRoutePolicy` and `ThrottlingExceptionRoutePolicy`), cannot wait for that pool: it would wait for itself until `shutdownAwaitTermination` and then drop the queued items anyway. In that case (second commit) the pool is only shut down without waiting: the queued exchanges run after the current one and complete, failed with a `RejectedExecutionException` that the consumer logs (or routed, if the consumer is started again first), so they are not left inflight. Stop time: a route stop now waits for the queued exchanges and no longer interrupts the running one right away, so a stop with a long-running exchange takes up to `shutdownAwaitTermination` (10 s) before the running exchange is interrupted (before: about 1 s, interrupted at once). A route that synchronously stops itself from one of its exchanges waits for itself until that timeout and is then stopped forcibly (before: forcibly at once); the upgrade note recommends stopping it asynchronously, as for `onCompletion().parallelProcessing()`. Upgrade guide note added. The defect was found with a TLA+ model of the subscriber (onNext, refill and the callback as locked sections), the pool and route stop/start: "every item taken from the stream is routed" is violated in 8 steps, and "a restarted idle consumer whose publisher has items left has requested some" in 19 steps. The model also rejected the first fix (drain after `super.doStop()`); the fix holds with one and two stop/start cycles. The model assumes the drain ends within the timeout and the stop comes from another thread; with the stop issued on the consumer's own thread (a route policy) the first commit alone deadlocks in the model (in Java: waits until the timeout, then drops), and the second commit holds (nothing dropped, inflight accounted, consumes again after a restart). Tests: new `ConsumerStopTest` (latches, Awaitility, no sleeps). Without the main-code change: ``` ConsumerStopTest.testStopProcessesTheExchangesTakenFromTheStream:53 The exchanges taken from the stream before the stop must be processed ==> expected: <[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]> but was: <[]> ConsumerStopTest.testConsumesAgainAfterRestart:64 » ConditionTimeout ... The route must consume again after its restart ==> expected: <true> but was: <false> within 10 seconds. ConsumerStopTest.testStopFromTheRouteDoesNotWaitForItself:92 » ConditionTimeout ... was not fulfilled within 10 seconds. ``` (the last one: the queued exchanges stay inflight). With the first commit only, `testStopFromTheRouteDoesNotWaitForItself:88 The route policy must be able to stop the consumer` fails (the stop waits for itself). The control `testConsumesAgainAfterRestartWithoutQueuedExchanges` passes before and after. With the change the camel-reactive-streams (69), camel-reactor (25) and camel-rxjava (25) tests pass. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested the affected modules, including the formatter and import-sort plugins. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commits carry a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
