You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/07 00:06:54 UTC

[PR] CAMEL-25350: camel-reactive-streams - complete the exchanges of a batch that the subscriber cancels before they are sent [camel]

allthingssecurity opened a new pull request, #27470:
URL: https://github.com/apache/camel/pull/27470

   # Description
   
   [CAMEL-25350](https://issues.apache.org/jira/browse/CAMEL-25350)
   
   `CamelSubscription.flush()` moves up to the requested number of exchanges from the buffer to a local queue, then sends them one by one and stops when the subscription was cancelled. The exchanges still in the local queue were dropped: they are no longer in the buffer, so `cancel()` does not discard them, and their dispatch callback is never called. The producer of `to("reactive-streams:x")` completes its exchange only from that callback, so a synchronous caller blocks forever and the route keeps the exchanges inflight (a graceful shutdown waits for them until its timeout).
   
   It happens whenever a subscriber that requested several exchanges cancels while a batch of more than one is being sent, in its `onNext` (Reactor `next()`, `takeWhile`, a subscriber that has seen enough) or from another thread. With `Flux.from(camel.fromStream("numbers", Integer.class)).next()` and 8 exchanges sent at once with `asyncSend`, 194 of 200 such subscriptions left exchanges that never completed (1037 in total) before this change, none after (a measurement, not a committed test).
   
   This change takes the exchanges from the local queue one at a time and, when the loop stops because of the cancel, hands the ones not sent to `discardBuffer`, as `cancel()` does with the buffer: they fail with the same `IllegalStateException` instead of never completing. A request of zero or less (rule 3.9 of the specification, the subscription ends with an error) left the buffered exchanges behind in the same way; they are now discarded too.
   
   No upgrade-guide note: the exchanges concerned never completed before, they now fail with the exception that `cancel()` already uses for the buffered ones.
   
   The defect was found with a TLA+ model of `CamelSubscription` (publish, request, checkAndFlush, the flush task with one step per locked section and per `onNext`, cancel in `onNext` or from another thread): "a published exchange that was not called back is held in the buffer, the sending queue, or being delivered or discarded" is violated in 14 steps (two exchanges buffered, request, the flush takes both, cancel in the first `onNext`, the loop breaks). It holds without a cancel, and with the fix.
   
   Tests: new `CancelSubscriptionTest` (Awaitility on the buffer size of the subscription, no sleeps). Without the main-code change:
   ```
   CancelSubscriptionTest.testCancelInOnNextCompletesTheOtherExchanges:56 ยป ConditionTimeout ... Exchange 2 never completed ==> expected: <true> but was: <false> within 5 seconds.
   CancelSubscriptionTest.testNonPositiveRequestCompletesTheBufferedExchanges ยป ConditionTimeout ... A buffered exchange never completed ==> expected: <true> but was: <false> within 5 seconds.
   ```
   With the change the camel-reactive-streams tests pass (67).
   
   Not changed: a subscriber whose `onNext` throws (forbidden by rule 2.13) still ends the flush task, leaving the rest of its batch and the subscription's `sending` flag behind (the `TODO` in `flush()`); that needs a decision on how to signal it and is left for a separate issue.
   
   # 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 module, 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 commit carries 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: commits-unsubscribe@camel.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] CAMEL-25350: camel-reactive-streams - complete the exchanges of a batch that the subscriber cancels before they are sent [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27470:
URL: https://github.com/apache/camel/pull/27470#issuecomment-6032347102

   <!-- ci-tested-modules -->
   :test_tube: **CI tested the following changed modules:**
   
   - `components/camel-reactive-streams`
   
   ---
   
   <details><summary>:microscope: Scalpel shadow comparison โ€” Scalpel: 11 of 699 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/engine/CamelSubscription.java, components/camel-reactive-streams/src/test/java/org/apache/camel/component/reactive/streams/CancelSubscriptionTest.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, 7m 17s total)</summary>
   
   **Total reactor time:** 7m 17s
   
   | Module | Duration | Status |
   | --- | --- | --- |
   | Camel :: Launcher | 50.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: TUI | 42.7s | SUCCESS |
   | Camel :: Reactive Streams | 38.3s | SUCCESS |
   | Camel :: Reactor | 37.9s | SUCCESS |
   | Camel :: RxJava | 37.8s | SUCCESS |
   | Camel :: JBang :: MCP | 36.9s | SUCCESS |
   | Camel :: Component DSL | 22.2s | SUCCESS |
   | Camel :: Catalog :: Camel Catalog | 20.1s | SUCCESS |
   | Camel :: YAML DSL :: Validator | 18.5s | SUCCESS |
   | Camel :: JBang :: Plugin :: Validate | 17.9s | SUCCESS |
   | Camel :: YAML DSL | 17.3s | SUCCESS |
   | Camel :: JBang :: Plugin :: Testing | 12.6s | SUCCESS |
   | Camel :: Docs | 11.1s | SUCCESS |
   | Camel :: JBang :: Plugin :: Kubernetes | 10.7s | SUCCESS |
   | Camel :: Kamelet Main | 10.7s | SUCCESS |
   | Camel :: YAML DSL :: Deserializers | 7.7s | SUCCESS |
   | Camel :: Catalog :: Camel Route Parser | 6.8s | SUCCESS |
   | Camel :: Catalog :: Camel Report Maven Plugin | 6.1s | SUCCESS |
   | Camel :: All Components Sync point | 5.4s | SUCCESS |
   | Camel :: YAML DSL :: Validator Maven Plugin | 4.7s | SUCCESS |
   | Camel :: Catalog :: Maven | 3.1s | SUCCESS |
   | Camel :: YAML DSL :: Maven Plugins | 2.8s | SUCCESS |
   | Camel :: Coverage | 2.4s | SUCCESS |
   | Camel :: Catalog :: Suggest (deprecated) | 2.2s | SUCCESS |
   | Camel :: Assembly | 1.8s | SUCCESS |
   | Camel :: JBang :: Plugin :: Edit | 1.7s | SUCCESS |
   | Camel :: JBang :: Plugin :: Generate | 1.3s | SUCCESS |
   | Camel :: Catalog :: Console | 1.0s | SUCCESS |
   | Camel :: Catalog :: Dummy Component | 1.0s | SUCCESS |
   | Camel :: JBang :: Integration tests | 0.9s | SUCCESS |
   | Camel :: JBang :: Main | 0.8s | SUCCESS |
   | Camel :: Endpoint DSL :: Support | 0.7s | SUCCESS |
   | Camel :: JBang :: Plugin :: Route Parser | 0.7s | SUCCESS |
   | Camel :: Launcher :: Container | 0.7s | SUCCESS |
   | Camel :: JBang :: Plugin :: MCP | 0.6s | SUCCESS |
   | Camel :: Endpoint DSL | n/a |  |
   | Camel :: Integration Tests | n/a |  |
   | Camel :: JBang :: Core | n/a |  |
   
   **Top 20 slowest modules:**
   - `Camel :: Launcher` (50.1s)
   - `Camel :: JBang :: Plugin :: TUI` (42.7s)
   - `Camel :: Reactive Streams` (38.3s)
   - `Camel :: Reactor` (37.9s)
   - `Camel :: RxJava` (37.8s)
   - `Camel :: JBang :: MCP` (36.9s)
   - `Camel :: Component DSL` (22.2s)
   - `Camel :: Catalog :: Camel Catalog` (20.1s)
   - `Camel :: YAML DSL :: Validator` (18.5s)
   - `Camel :: JBang :: Plugin :: Validate` (17.9s)
   - `Camel :: YAML DSL` (17.3s)
   - `Camel :: JBang :: Plugin :: Testing` (12.6s)
   - `Camel :: Docs` (11.1s)
   - `Camel :: JBang :: Plugin :: Kubernetes` (10.7s)
   - `Camel :: Kamelet Main` (10.7s)
   - `Camel :: YAML DSL :: Deserializers` (7.7s)
   - `Camel :: Catalog :: Camel Route Parser` (6.8s)
   - `Camel :: Catalog :: Camel Report Maven Plugin` (6.1s)
   - `Camel :: All Components Sync point` (5.4s)
   - `Camel :: YAML DSL :: Validator Maven Plugin` (4.7s)
   
   </details>
   
   ---
   :gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37550237521)


-- 
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-25350: camel-reactive-streams - complete the exchanges of a batch that the subscriber cancels before they are sent [camel]

Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27470:
URL: https://github.com/apache/camel/pull/27470


-- 
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-25350: camel-reactive-streams - complete the exchanges of a batch that the subscriber cancels before they are sent [camel]

Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on PR #27470:
URL: https://github.com/apache/camel/pull/27470#issuecomment-6033226491

   Thanks @davsclaus. I checked the follow-up on `main` (fb86fbed28dc) and you are right. It happens even without a race: the completion branch of `checkAndFlush()` discards the live buffer without clearing it, so a subscriber that calls `cancel()` after `onComplete` makes `cancel()` discard the same exchanges again. In a quick unit test on `CamelSubscription` (two buffered exchanges, `signalCompletion()` then `cancel()`), the callbacks run 4 times instead of 2. For the producer that means `callback.done(false)` twice.
   
   It is limited in practice: `signalCompletion()` is only called from `CamelPublisher.close()`, and `DefaultCamelReactiveStreamsService` never closes its publishers. The fix would be to copy and clear the buffer under the mutex in that branch, as `cancel()` does. I'll keep it out of this PR and propose a separate ticket for it.
   
   _Claude Code on behalf of allthingssecurity_
   


-- 
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-25350: camel-reactive-streams - complete the exchanges of a batch that the subscriber cancels before they are sent [camel]

Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27470:
URL: https://github.com/apache/camel/pull/27470#issuecomment-6031972870

   :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