You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "gnodet (via GitHub)" <gi...@apache.org> on 2026/10/06 10:02:17 UTC
[PR] CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges [camel]
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27434:
URL: https://github.com/apache/camel/pull/27434#issuecomment-6014420736
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-reactive-streams`
:repeat: **1 test passed only after a retry on JDK 25** (1 retried attempt)
<details><summary>Recovered flaky tests on JDK 25 (1)</summary>
| Module | Test | Failed attempts | First failure |
| --- | --- | --- | --- |
| `components/camel-reactor` | `ReactorStreamsServiceTest.testToFunction` | 1 | Timeout on blocking read for 30000000000 NANOSECONDS |
> :information_source: These tests did **not** fail the build. Surefire retried them and they passed.
> Retries are enabled project-wide by `surefire.rerunFailingTestsCount` in `parent/pom.xml`.
</details>
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 11 of 698 tested, 27 compile-only — current: 11 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **11 affected modules** (current approach: 11).
**Skip-tests mode would test 11 modules** (1 direct + 10 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (11)</summary>
- `camel-jbang-mcp` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` ← downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` ← downstream of org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` ← downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` ← downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` ← downstream of org.apache.camel:camel-launcher
- `camel-reactive-streams` ← components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java, components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/ConsumerStopTest.java
- `camel-reactor` ← downstream of org.apache.camel:camel-reactive-streams
- `camel-rxjava` ← downstream of org.apache.camel:camel-reactive-streams
- `camel-yaml-dsl-validator` ← downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← downstream of org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (38 modules, 6m 46s total)</summary>
**Total reactor time:** 6m 46s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 49.8s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 42.6s | SUCCESS |
| Camel :: JBang :: MCP | 39.5s | SUCCESS |
| Camel :: RxJava | 38.6s | SUCCESS |
| Camel :: Reactive Streams | 38.4s | SUCCESS |
| Camel :: Component DSL | 24.3s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 23.1s | SUCCESS |
| Camel :: YAML DSL :: Validator | 19.9s | SUCCESS |
| Camel :: YAML DSL | 18.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 18.5s | SUCCESS |
| Camel :: Docs | 15.1s | SUCCESS |
| Camel :: Kamelet Main | 11.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 8.3s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 8.2s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 7.7s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 6.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 5.3s | SUCCESS |
| Camel :: All Components Sync point | 4.9s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 4.2s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.1s | SUCCESS |
| Camel :: Catalog :: Maven | 3.0s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.4s | SUCCESS |
| Camel :: Assembly | 1.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.4s | SUCCESS |
| Camel :: Coverage | 1.3s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.1s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: JBang :: Integration tests | 0.8s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.7s | SUCCESS |
| Camel :: JBang :: Main | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.6s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.6s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: Reactor | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (49.8s)
- `Camel :: JBang :: Plugin :: TUI` (42.6s)
- `Camel :: JBang :: MCP` (39.5s)
- `Camel :: RxJava` (38.6s)
- `Camel :: Reactive Streams` (38.4s)
- `Camel :: Component DSL` (24.3s)
- `Camel :: Catalog :: Camel Catalog` (23.1s)
- `Camel :: YAML DSL :: Validator` (19.9s)
- `Camel :: YAML DSL` (18.7s)
- `Camel :: JBang :: Plugin :: Kubernetes` (18.5s)
- `Camel :: Docs` (15.1s)
- `Camel :: Kamelet Main` (11.5s)
- `Camel :: JBang :: Plugin :: Testing` (8.3s)
- `Camel :: YAML DSL :: Deserializers` (8.2s)
- `Camel :: Catalog :: Camel Route Parser` (7.7s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (6.5s)
- `Camel :: JBang :: Plugin :: Validate` (5.3s)
- `Camel :: All Components Sync point` (4.9s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (4.2s)
- `Camel :: YAML DSL :: Maven Plugins` (3.1s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37447072684)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges [camel]
Posted by "gnodet (via GitHub)" <gi...@apache.org>.
gnodet closed pull request #27434: CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges
URL: https://github.com/apache/camel/pull/27434
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27434:
URL: https://github.com/apache/camel/pull/27434#issuecomment-6013921777
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25351: camel-reactive-streams - fix consumer stop dropping queued exchanges [camel]
Posted by "gnodet (via GitHub)" <gi...@apache.org>.
gnodet commented on PR #27434:
URL: https://github.com/apache/camel/pull/27434#issuecomment-6013956323
<!-- camel-backport-bot -->
### 🔄 Backport Bot
This bugfix targets `main` and may need porting to:
- `camel-4.22.x`
Add a `port/<branch>` label to automatically create a backport PR for that branch.
Backport PRs will be created automatically on merge (once a port label is added). Comment `/port` to create them immediately.
ℹ️ If you push additional commits after `/port`, use `/port` again to update the port PRs.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org