allthingssecurity opened a new pull request, #28785: URL: https://github.com/apache/camel/pull/28785
# Description [CAMEL-26430](https://issues.apache.org/jira/browse/CAMEL-26430) Follow-up of CAMEL-26327 (#28750), asked for in its review. When the error handler redelivers an exchange and the processor throws from `process(exchange, callback)` instead of setting the exception on the exchange, the exchange never completes. A synchronous `template.send` or a consumer thread blocks forever, and an asynchronous caller never gets its callback: ```java errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0)); from("direct:start").process(new AsyncProcessorSupport() { int calls; public boolean process(Exchange exchange, AsyncCallback callback) { if (++calls == 1) { exchange.setException(new IOException("first")); // the delivery fails as it should callback.done(true); return true; } throw new IllegalStateException("second"); // the redelivery throws } }); template.send("direct:start", "Hello"); // never returns ``` Cause: `RedeliveryErrorHandler.RedeliveryTask.run()` wraps its work in a try/catch, but `RedeliveryTask.redeliver()` does not. A redelivery runs as a task of its own on the reactive executor (`reactiveExecutor.schedule(this::redeliver)` for an immediate redelivery and after a synchronous delay, and from the scheduled executor for an asynchronous delayed redelivery). An exception thrown there reaches `DefaultReactiveExecutor`, which logs `Error executing reactive work due to ... This exception is ignored.`, and the callback of the exchange is never called. A plain `Processor` that throws does not hang, because the converter to `AsyncProcessor` catches its exception and sets it on the exchange. CAMEL-26327 (a routing slip step whose producer was gone, NPE on redelivery) is such a hang. A `StepHook` that throws does not hang, as `StepHelper` already catches it (CAMEL-26338), and a failing `onRedelivery` processor is caught too. ## The choice: loop back to the error handler, not call the callback directly The JIRA leaves two options: call the callback directly, as `run()` does, or loop back so that the error handler handles the new failure. This PR loops back, because: - On the redelivery path, every other failure already loops back: an `AsyncProcessor` that sets the exception on the exchange, a plain `Processor` that throws (its converter sets the exception), and a failing `onRedelivery` processor (`redeliver()` already schedules `this` for it). An `AsyncProcessor` that throws is the same failure, delivered differently, so it gets the same handling: the exception policy is chosen for the new exception, the redelivery counter goes up, the `onRedelivery` processor runs before the next attempt, and the error handler redelivers again or gives up (dead letter channel, handled or continued `onException`, or the failure goes to the caller). - It cannot loop forever: each loop is one more delivery attempt and counts against `maximumRedeliveries`. Only `maximumRedeliveries(-1)` retries without end, as it does for any processor that keeps failing. - Calling the callback directly would let the exchange fail past a dead letter channel that is configured to take every failure, and past a handled `onException` for that exception type (a negative control with the direct callback fails those tests). The takeover is scheduled on the reactive executor, not run inside the catch. A processor can hand its callback to another processor and then throw. When that work is already queued on the reactive executor when the processor throws (such as the steps of a route behind a `direct` producer, each wrapped in its error handler), it runs after `redeliver()` returns, and its callback completed the exchange on main (the exception was only logged). It still does: the takeover is queued behind that work, so the callback wins (test `controlQueuedCallbackBeforeTheProcessorThrows`, with a two-step inner route; with the takeover inside the catch, that test fails). Work that is not queued there yet when the processor throws (another thread, or a route with `noErrorHandler()` whose later steps are queued only after the takeover) loses to the takeover: see the behaviour change below. ## Exactly once The callback given to the processor and the takeover share one `AtomicBoolean` per redelivery: whichever sets it first continues the exchange, and the other does nothing. The catch and the takeover read no task field before they have won, because when the callback won, the task may already be released (pooled tasks are reset on release); the exchange and its id are read into locals while the redelivery still owns the task. | Path in `redeliver()` | Who continues the exchange | What the other one does | |---|---|---| | `prepareExchangeForRedelivery` throws (no callback handed out yet) | the takeover: exception on the exchange, back to the error handler | nothing to do | | the `onRedelivery` processor fails (caught inside, unchanged) | `schedule(this)`, return | nothing to do | | the exchange is marked to stop (the flag line is new) | the flag is set, the task is released, `schedule(callback)` | nothing to do | | the processor calls the callback, sync or async, and does not throw (unchanged) | the callback (as before: done, or loop back) | nothing to do | | the processor throws and never calls the callback (the bug) | the takeover | nothing to do | | the processor calls the callback, then throws | the callback | the catch sees the flag set and rethrows; the reactive executor logs it, as before | | the processor's callback is queued on the same reactive worker, then the processor throws | the queued callback (it runs first) | the takeover loses and logs one WARN with the exception, as the reactive executor did before | | the processor hands the callback to another thread, then throws; the other thread calls it after the takeover | the takeover | the late callback loses, logs one WARN and returns | | as above, both at the same time | the one whose `compareAndSet` wins | the loser logs and returns | | the callback itself throws while the processor calls it synchronously | the callback (it set the flag first) | the exception reaches the catch, which rethrows, as before | After the takeover, `run()` checks `isRunAllowed()` as for any failure: on a forced shutdown, or when redelivery is not allowed while stopping (and the error handler is not a dead letter channel, which keeps running while Camel prepares to stop), the exchange completes with the thrown exception; otherwise the error handler goes on (by default `allowRedeliveryWhileStopping` is true). The catch takes `Exception`, as `run()` and `DefaultReactiveExecutor` do. An `Error` thrown from a processor is not changed by this PR. ## Changes - `RedeliveryTask.redeliver()` runs its old body (now `doRedeliver`) in a try/catch with the shared flag above, and the takeover sets the exception and calls `run()`. The cost is one `AtomicBoolean` per redelivery, only on the redelivery path; the first delivery and the success path are unchanged. - One behaviour change besides the hang: a processor that hands its callback on and then throws (an `AsyncProcessor` must not throw from `process`, but nothing stopped it), when that callback is not already queued on the reactive executor: another thread, or a route behind a `direct` producer with `noErrorHandler()` and more than one step. Before, the exchange waited for that callback, and the exception was only logged. Now the thrown failure is handled at once (redelivered, or the dead letter channel, `onException` or the caller), and the callback, when it comes later, is ignored with a WARN. The error handler cannot tell such a callback from one that never comes, which is the hang. Meanwhile the other thread may still write to the exchange. No upgrade guide entry for this: it needs a processor that breaks the `AsyncProcessor` contract, and the first delivery (`run()`) already treats such a throw as the failure of the exchange. - No upgrade guide entry for the fix itself: an exchange whose callback never came hung forever, so no route can depend on it. Not changed, for a follow-up JIRA I will file (all on main as well): - An error handler without redelivery runs the step through `SimpleTask`, whose `run()` has no try/catch either: an `AsyncProcessor` that throws there hangs the exchange with the default error handler (CAMEL-26301, the tokenizer, was this path). `SimpleTask` is the success path of every step, so it needs its own change and tests. - The first delivery of `RedeliveryTask` (`run()`) calls the callback directly when the processor throws, so a throw on the first attempt skips redelivery, the dead letter channel and `onException`, while the same throw on a redelivery now goes through them. - `run()` and the "simple delivery" callback in `doRun` have no flag like the one above: a processor that hands its callback to another thread and then throws on the first delivery gets the exchange completed by `run()`'s catch and continued a second time by the callback (two `ExchangeCompleted`/`ExchangeFailed` events; with pooled tasks the late callback works on a reset task). ## Tests New `RedeliveryProcessorThrowsTest` (camel-core, package `processor.errorhandler`, 13 tests, JUnit assertions as in the package), and `RedeliveryProcessorThrowsPooledTest`, the same tests with pooled exchanges (so the redelivery tasks are pooled and reset when released). The processor fails the first call as it should and throws on the second; every send is bounded with `asyncSend(...).get(10, SECONDS)`, no sleeps and no assert periods: - immediate, synchronous delayed, asynchronous delayed redelivery, asynchronous delayed with `allowRedeliveryWhileStopping(false)` (it then wakes up to check for a stop before it redelivers, a different path to `redeliver()`), and a transacted exchange with immediate and synchronous delayed redelivery: the send completes with the thrown exception, the processor was called twice; - the thrown failure is redelivered: with 2 redeliveries the third call succeeds, `CamelRedeliveryCounter` is 2; with an `onRedelivery` processor, it runs before each of the 2 redeliveries; - a dead letter channel (with asynchronous delayed redelivery) gets the thrown failure (`CamelExceptionCaught` is the thrown exception), and a handled `onException` for the thrown exception type handles it (the policy is chosen again for the new failure); - control: the processor calls the callback and then throws: the exchange completes once; - control: the processor hands the callback to a `direct` route and then throws: the queued callback completes the exchange with the route's reply, as on main; - the processor hands the callback to another thread and throws; after the exchange completed, that thread calls the callback: it is ignored, the exchange is done once (counted with an `EventNotifier` on `ExchangeCompletedEvent` / `ExchangeFailedEvent`), the dead letter channel got it once. This test pins the behaviour change above: on main the exchange waits for that callback, which the test releases only after the send returned, so main times out. Without the fix, 11 of the 13 tests time out in each class; the two controls pass on main as well. Mutants of the fix, each failing at least one test: taking over inside the catch (fails the queued-callback control), no flag in the callback (fails the done-once count), calling the callback directly instead of looping back (fails the dead letter channel, `onException`, redelivery and `onRedelivery` tests), no flag check in the takeover. - Each new class 20 times in one JVM: 260 tests each, 0 failures. - camel-core `org.apache.camel.processor.**` (with `errorhandler`, `onexception`, `exceptionpolicy` and the other subpackages) plus the doTry/doCatch/doFinally, redelivery, error handler and exception tests of `org.apache.camel.issues`: 1442 classes, 2956 tests, 0 failures, 24 skipped. Compared per class with the same run on main 0fd7d9abf9c2 (1440 classes, 2930 tests, 0 failures): identical except the two new classes. - Together with CAMEL-26327 (#28750), CAMEL-26333 (#28756) and CAMEL-26434 of this batch on one tree: the tests of the four branches and the doTry/doCatch/doFinally, routing slip, recipient list, dynamic router, redelivery, dead letter channel, error handler, onException and onCompletion tests of camel-core pass (1042 tests, 0 failures, 10 skipped). The branch merges cleanly with each of them. - 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-26430) filed for the change (usually before you start working on it): [CAMEL-26430](https://issues.apache.org/jira/browse/CAMEL-26430). # 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]
