gnodet opened a new pull request, #27434:
URL: https://github.com/apache/camel/pull/27434
## Summary
Fix `ReactiveStreamsConsumer.doStop()` so that exchanges already taken from
the stream are not dropped on route stop, and the consumer can resume
requesting items after a restart.
## Root Cause
`doStop()` had two interacting defects:
**Defect 1 — dropped queued exchanges.**
`super.doStop()` was invoked first, which stops the route processor (error
handler). Then `shutdownNow()` was called on the executor, which discarded
pending tasks without their `AsyncCallback` ever firing.
**Defect 2 — permanently inflated `inflightCount`.**
Because the callbacks for the dropped exchanges never fired, `inflightCount`
in `ReactiveStreamsCamelSubscriber` was permanently incremented. After a route
restart, `refill()` computed:
```
newRequest = max - requested - inflightCount
= 10 - 0 - 9 = 1
```
`minRequests = max × (1 − lowWatermark) = 10 × 0.75 = 7.5 ≈ 8`
Since `1 < 8`, no new requests were issued — the restarted route consumed
nothing.
The defect affects the default engine and the `camel-reactor` /
`camel-rxjava` integrations, which use the same consumer and subscriber classes.
## Fix
Three changes, in order:
1. **Detach consumer first** — `service.detachCamelConsumer()` is moved to
the top of `doStop()`. This stops the subscriber from handing new items to the
executor immediately, while leaving items already queued in the executor
untouched.
2. **Replace `shutdownNow()` with `shutdownGraceful()`** — the executor
drains its queue, all pending exchange tasks complete, and every
`AsyncCallback` fires. `inflightCount` returns to zero before `doStop()`
returns.
3. **Call `super.doStop()` last** — the route processor is only stopped
after the executor is fully drained, so no in-flight exchange is rejected by
the error handler.
## Test
`ConsumerStopTest` (new):
| Test | What it verifies |
|------|-----------------|
| `testStopProcessesTheExchangesTakenFromTheStream` | 10 items subscribed;
first exchange holds a latch so 9 queue in the executor; route is stopped — all
10 must be processed (not dropped) |
| `testConsumesAgainAfterRestart` | Same setup then route is stopped and
restarted — fresh publisher items must be consumed (inflightCount must be 0
after stop) |
| `testConsumesAgainAfterRestartWithoutQueuedExchanges` | Clean stop
(nothing queued) then restart — basic restart must still work (control) |
| `testStopFromTheRouteDoesNotWaitForItself` | Asynchronous stop from inside
a processor must not deadlock |
All 4 tests fail on the unfixed code, all 4 pass with the fix. The full
module test suite (69 tests) is green.
---
_Hermes Agent (Claude Sonnet 4.6) on behalf of Guillaume Nodet_
--
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]
