allthingssecurity opened a new pull request, #29040: URL: https://github.com/apache/camel/pull/29040
# Description [CAMEL-26446](https://issues.apache.org/jira/browse/CAMEL-26446) Follow-up of CAMEL-26430 (#28785, merged: a redelivery that throws) and CAMEL-26445 (the `SimpleTask`), as asked in the review of #28785: the first delivery of `RedeliveryTask` now handles a throw the same way. `RedeliveryErrorHandler` runs a step through its `RedeliveryTask` when its error handler can redeliver, has an exception policy (`onException`) or an `onPrepareFailure` processor, or is a dead letter channel. The first delivery of the step is the "simple delivery" branch of `doRun()`. When the processor (such as an `AsyncProcessor`) throws from `process(exchange, callback)` there instead of setting the exception on the exchange, the exception went to the catch of `run()`, which set it on the exchange, released the task and called the callback directly: ```java errorHandler(deadLetterChannel("mock:dead").maximumRedeliveries(2)); from("direct:start") .process(new AsyncProcessorSupport() { @Override public boolean process(Exchange exchange, AsyncCallback callback) { throw new IllegalStateException("thrown"); // instead of exchange.setException(...) + callback.done(true) } }) .to("mock:result"); // on main: the exchange fails at once with the exception: no redelivery, nothing in mock:dead, no onException ``` - the throw skipped redelivery, the dead letter channel and `onException`, while the same failure set on the exchange, or thrown by a plain `Processor` (whose converter catches it), goes through them; - the catch had no exactly-once guard: a processor that called its callback (or handed it to another thread) and then threw got the exchange completed twice: once by the catch with the exception, once by the callback (with pooled tasks, the catch failed with a `NullPointerException` on the released task instead). ## The fix The simple delivery moves into `deliver()`, with a try/catch around the call of the processor, the same rule as CAMEL-26430 and CAMEL-26445: - a throw before the callback was called: the failure is taken over through the reactive executor, after the work already queued there, so a callback that the processor handed to queued work still comes first (such as the reply of a route behind a `direct` producer with the default error handler, whose steps each schedule the next one there; a route with `noErrorHandler()` loses to the takeover, and its callback is then ignored). The takeover sets the exception on the exchange and continues exactly as the callback does when the processor sets the exception (the same tail: `isDone`, else loop back to `run()`): the exception policy is chosen, `onExceptionOccurred` runs, and the exchange is redelivered or goes to the dead letter channel / `onException`; - a throw after the callback was called: the callback continued the exchange, so the exception is only logged at WARN (before, the catch failed and completed the exchange a second time); - exactly once: the task keeps the number of the first delivery whose callback is still to come in one `volatile int` (`awaited`, an `AtomicIntegerFieldUpdater`, as the `SimpleTask` fix does), and each first delivery gets a number (`deliveries`, per task, never reset, so a callback of an earlier exchange of a pooled task never matches). The callback and the takeover `compareAndSet` it to 0; only the winner continues the exchange (both through the same tail as before: done, or loop back to `run()`), the loser logs one WARN. The callback is a lambda, as before (it captures the delivery number now); on the success path the cost is one ordered write before the processor is called and one uncontended `compareAndSet` when its callback comes. Nothing else is allocated unless the processor throws. - the WARNs name the processor, not the exchange id: the id is created on first use (a split or multicast copy may not have one yet), and on these paths the exchange may be in use on another thread, or, pooled, by another exchange, so reading it there could create or read the id of the wrong exchange. The rest of `run()`'s catch is unchanged (failures of the error handler's own steps, such as `handleException` or the dead letter channel). ### Relation to #28785 (CAMEL-26430) and the `SimpleTask` PR (CAMEL-26445) #28785 changed `redeliver()` and is on main. One difference: its takeover calls `run()` directly, while the takeover here continues through the same tail as the callback (`isDone`, else schedule `this`), which is the path of a failure set on the exchange; the two differ only for an exchange marked interrupted, failure handled or exhausted, where the tail here matches the callback. The `SimpleTask` PR changes `SimpleTask`; this one changes the simple delivery of `RedeliveryTask.doRun()` and adds two fields to `RedeliveryTask`. Each keeps its own small takeover (a few lines each, with different state: an `AtomicBoolean` per redelivery in #28785, a call number on the task here and in `SimpleTask`); CAMEL-24980 proposes to share the failure handling of the tasks, which could fold them together later. This branch is on current main (with #28785), and merges cleanly with the `SimpleTask` PR (`git merge-tree`); the tests of all of them pass together. With #28785 on main, a first delivery that throws followed by a redelivery that throws fails with the exception of the redelivery (test `throwsOnEveryAttempt`). Not changed (on main as well): an `AsyncProcessor` that throws inside a `doTry` block or an `onException` still hangs the exchange: the steps of these blocks have no error handler, so the throw goes from `CamelInternalProcessor` up to the reactive work of `Pipeline`, which drops it (the same with all four PRs merged). Follow-up: CAMEL-26464. Camel's own processors throw from `process(exchange, callback)` only in their "not started" guards (`RecipientList`, `WireTapProcessor`, `ThreadsProcessor`); under a redelivering error handler such a failure now goes through it as well. Backport: main only (a behaviour change, with an upgrade-guide entry; it builds on `processOutput` of CAMEL-26338 and on CAMEL-26430, which are on main only). ## Behaviour changes - An `AsyncProcessor` (used with `process(...)`; endpoints in `to(...)` are not affected, as `SendProcessor` catches the producer's exception) that throws on the first attempt, under an error handler with redelivery, `onException`, `onPrepareFailure` or a dead letter channel: the error handler now handles the failure like any other failure of the step (redelivers it, and when exhausted the matching `onException` or the dead letter channel handles it). Before, the exchange failed at once, so the processor is now called again on redelivery, a dead letter channel now takes the message (the exchange ends handled instead of failing back to the consumer), and the error handler logs the exhausted failure. Upgrade guide entry (4.24). The way to not redeliver it is an `onException` for its exception type with `maximumRedeliveries(0)` (pinned by a test; it then still goes to the dead letter channel / `onException`, as the same failure set on the exchange does). There is no option to let it skip the dead letter channel or `onException` as before; the guide says so. - A processor that hands its callback to another thread and then throws: the thrown failure is handled right after the work already queued on the reactive executor, and the callback, when it comes later, is ignored with a WARN (a callback that comes before the takeover wins, and the thrown exception is ignored with a WARN) (before, the catch completed the exchange and the callback continued it again). The error handler cannot tell such a callback from one that never comes. The other thread may still write to the exchange meanwhile, as with CAMEL-26430. - A callback called twice by a processor: the second call is ignored with a WARN (before, it continued the exchange a second time, or, with pooled exchanges, failed with a `NullPointerException` on the released task). ## Tests New `RedeliveryFirstDeliveryProcessorThrowsTest` (camel-core, `processor.errorhandler`, next to the tests of CAMEL-26430 and CAMEL-26445, JUnit assertions as in the package, 19 tests) and `RedeliveryFirstDeliveryProcessorThrowsPooledTest` (the same with pooled exchanges, so pooled redelivery tasks). Sends that could hang are bounded (`asyncSend(...).get(10, SECONDS)`), no sleeps: - the thrown failure is redelivered: immediate, synchronous delayed, asynchronous delayed, and transacted redelivery (redelivery counter 1, `mock:result` once); - exhausted: the exchange fails with the failure of the redelivery, also when the redelivery throws too; with a dead letter channel that redelivers, the dead letter channel gets the failure of the redelivery; - the way to not redeliver that the upgrade guide gives (`onException(IllegalStateException.class).maximumRedeliveries(0)` with a redelivering dead letter channel): called once, the dead letter channel gets it; - dead letter channel: it gets the thrown exception (`EXCEPTION_CAUGHT`); - `onException(IllegalStateException.class).handled(true)` handles it; - the next exchange after a takeover is processed normally; - the processor calls the callback, then throws: the exchange completes once, nothing in the dead letter channel; - the processor hands the exchange and callback to a `direct` route (default error handler), then throws: the route's reply wins, once; - the processor hands the callback to another thread and throws; the late callback is ignored, the exchange is done once (counted with an `EventNotifier`); - as above while the next exchange waits for its own first delivery: the late callback does not continue it; - the callback called twice: done once; - the callback and the throw at the same time (stress test): 500 exchanges whose processor hands its callback to a pool of two threads, which calls it at once, and throws (the takeover wins nearly always); each exchange is done once, `mock:result` + `mock:dead` = 500; - controls (pass on main too): a processor that sets the exception on the exchange goes to the dead letter channel; a plain `Processor` that throws goes to the dead letter channel. With main's `RedeliveryErrorHandler` (compiled over the branch), 17 of the 19 tests fail (16 pooled): the thrown failure is not redelivered and does not reach the dead letter channel or `onException`, and the exchange is completed twice; the 2 controls pass (pooled, also the callback-before-throw test, where the catch of main fails with a `NullPointerException` on the released task and the exchange completes once). Mutants: - the takeover run inside the catch instead of through the reactive executor: the `direct` route test fails (both classes); - no `compareAndSet` in the callback: the late callback, the double callback and the stress test fail (and, pooled, the late callback continues the next exchange); - the delivery number not kept across uses of a pooled task (always 1): the pooled late-callback test fails (the late callback of the first exchange continues the next one). - Both new classes 20 times in one JVM: 760 tests, 0 failures. - camel-core `org.apache.camel.processor.**` plus the error handler, redelivery, exception, doTry and doCatch tests of `org.apache.camel.issues`, with the new classes: 3013 tests, 0 failures, 24 skipped (before the review changes and the rebase); after them, `processor.errorhandler.**`, `processor.onexception.**` and the doTry/doCatch/doFinally tests: 318, 0 failures, 1 skipped; on current main, the tests of #28785 and of this PR: 58, 0 failures. - With #28785 (CAMEL-26430), the `SimpleTask` PR (CAMEL-26445) and the `NoErrorHandler` follow-up merged in (all merge cleanly with each other and with main): the tests of all four, 96 tests, 0 failures (before `throwsOnEveryAttempt` was added). - Formatter and import-sort: no changes. # 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-26446) filed for the change (usually before you start working on it): [CAMEL-26446](https://issues.apache.org/jira/browse/CAMEL-26446). # 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 the core modules and ran the tests above, including the formatter and import-sort plugins. No generated file changes. 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
