[jira] [Updated] (CAMEL-25350) camel-reactive-streams - exchanges of a batch never complete when the subscriber cancels while the batch is sent

["Claus Ibsen (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CAMEL-25350?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Claus Ibsen updated CAMEL-25350:
--------------------------------
    Fix Version/s: 4.23.0

> camel-reactive-streams - exchanges of a batch never complete when the subscriber cancels while the batch is sent
> ----------------------------------------------------------------------------------------------------------------
>
>                 Key: CAMEL-25350
>                 URL: https://issues.apache.org/jira/browse/CAMEL-25350
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-reactive-streams
>            Reporter: shashank
>            Assignee: shashank
>            Priority: Major
>             Fix For: 4.23.0
>
>
> {{CamelSubscription.flush()}} (default reactive streams engine) moves up to the requested number of exchanges from the buffer of a subscription to a local queue, then calls {{onNext}} for each and stops when the subscription was cancelled. The exchanges still in the local queue are then dropped: they are not in the buffer anymore, so {{cancel()}} does not discard them, and their dispatch callback is never called.
> The producer of {{to("reactive-streams:x")}} (also used by {{CamelReactiveStreamsService.from(uri)}}) completes its exchange only from that callback, so:
> * a synchronous caller (a consumer thread, {{ProducerTemplate.sendBody}}) blocks forever;
> * the exchanges stay inflight in the route, and a graceful shutdown waits for them until its timeout.
> It happens whenever a subscriber that has requested several exchanges cancels while a batch with more than one exchange is being sent: in its {{onNext}} (for example Reactor's {{next()}}, {{takeWhile}}, a subscriber that has seen enough) or from another thread (dispose, timeout). A batch holds several exchanges as soon as the producers publish faster than one {{onNext}} at a time, or exchanges were buffered while the subscriber had no demand.
> h3. Reproduction
> {{CancelSubscriptionTest}}: three exchanges are sent to {{direct:numbers -> reactive-streams:numbers}} while the subscriber has no demand; it then requests 3 and cancels in its first {{onNext}}. On main:
> {noformat}
> CancelSubscriptionTest.testCancelInOnNextCompletesTheOtherExchanges:56 ConditionTimeout
>   Exchange 2 never completed ==> expected: <true> but was: <false> within 5 seconds.
> {noformat}
> The control {{testDeliveredWithoutCancel}} passes. No sleeps (Awaitility on the buffer size of the subscription).
> How often it happens in practice (a measurement, not a committed test): with {{Flux.from(camel.fromStream("numbers", Integer.class)).next().subscribe()}} and 8 exchanges sent at once with {{ProducerTemplate.asyncSend}} to {{direct:numbers -> reactive-streams:numbers}}, 194 of 200 such subscriptions left exchanges that never completed on main (1037 exchanges in total); with the fix none.
> The defect was found with a TLA+ model of {{CamelSubscription}} (producers' {{publish}}, {{request}}, {{checkAndFlush}}, the flush task with one step per locked section and per {{onNext}}, {{cancel}} in {{onNext}} or from the subscriber's thread): "an exchange that was published and 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, the subscriber cancels in the first {{onNext}}, the loop breaks). Without a cancel it holds; with the fix it holds for both cancel modes, with 3 and 4 exchanges.
> h3. Proposed fix
> {{flush()}} takes the exchanges from its local queue one by one and, when it stops because the subscription was cancelled, passes the ones not sent to {{discardBuffer}}, as {{cancel()}} does with the buffer: their exchanges fail with the same {{IllegalStateException}} ("subscription cancelled") instead of never completing. No exchange is called back twice (the local queue and the buffer are disjoint).
> With the fix the camel-reactive-streams tests pass (67).
> Affected: 4.14.x, 4.18.x and main (same code, GitHub contents API); the loop dates from the component's first version.
> Duplicate check (2026-10-05, again by the review): JIRA component camel-reactive-streams (11 issues since 2017: CAMEL-23918 flaky tests, CAMEL-18212, CAMEL-15294, CAMEL-14219, CAMEL-11615, ...), text "CamelSubscription" (CAMEL-11140 uuid, CAMEL-11615, CAMEL-11124): none about this. GitHub pull requests "reactive-streams", "CamelSubscription": none (#24448 and #27012 only fixed flaky tests).
> _Filed with Claude Code on behalf of allthingssecurity._



--
This message was sent by Atlassian Jira
(v8.20.10#820010)