allthingssecurity opened a new pull request, #28598: URL: https://github.com/apache/camel/pull/28598
# Description [CAMEL-26331](https://issues.apache.org/jira/browse/CAMEL-26331), the first code step of [CAMEL-26330](https://issues.apache.org/jira/browse/CAMEL-26330) (camel-chaos), as in the design document `design/chaos.adoc` (#28539, "Hooks at all levels" and "Order of work"). Every send to an endpoint goes through one of six places that fire `ExchangeSendingEvent`, and each of them repeats the same code: fire the sending event, start a `StopWatch`, call the producer, fire the sent event. This PR moves that code into one helper, `SendHelper` (camel-support), and gives the helper a hook: a new SPI, `org.apache.camel.spi.SendHook`, installed with `ExtendedCamelContext#setSendHook`. The hook is general (it names nothing of chaos), so other developer tools and tests can use it. ```java public interface SendHook { long PROCEED = 0; long SKIP = -1; long beforeSend(Exchange exchange, Endpoint endpoint); // PROCEED, a delay in millis, or SKIP void afterSend(Exchange exchange, Endpoint endpoint); } ``` Before the producer, the hook can change the message, stop the send (it sets an exception on the exchange, and the producer is not called), skip it (the send completes without the producer, as if it had succeeded) or delay it. After the producer, it can change the reply. `SKIP` is not in the JIRA; chaos needs it for its `drop` fault, and interceptSendToEndpoint has the same idea (`skipSendToOriginalEndpoint`). **The six places** (the same set on main today, found with `git grep notifyExchangeSending`): | Place | Sends | Events | Hook | |---|---|---|---| | `SendProcessor` | `to` (singleton producer) | `SendHelper.notifySending/notifySent` | `SendHelper.process` around `producer.process` | | `DefaultProducerCache.doInAsyncProducer` | `toD`, `enrich`, `wireTap`, `routingSlip`, `dynamicRouter`, `to` (non-singleton producer), `ProducerTemplate.asyncSend` | same | the callers get the producer wrapped by `SendHelper.withSendHook`, so the hook runs where they call it | | `DefaultProducerCache.send` | `ProducerTemplate.send` | same | the producer passed to the shared internal processor is wrapped the same way | | `EventNotifierCallback` | `ProducerTemplate.asyncSend` (inside `doInAsyncProducer`) | same | none here: `doInAsyncProducer` already gave it the wrapped producer | | `MulticastProcessor` (`beforeSend`/`afterSend`) | `recipientList` | same | the producer of each recipient, inside the error handler of that recipient | | `EventNotifierProducer` | deferred producers (`@Produce`, `DeferServiceFactory`) | `SendHelper.send` (events + hook) | same | The events and the hook are two parts of the helper because two places fire the events once per send but call the producer inside an error handler of their own, which can call it again: the recipient list (an error handler per recipient) and the routing slip (an error handler per step). The hook runs where the producer is called, so for them it runs inside that error handler, on every attempt, as a failure of the producer would. In the `MulticastProcessor`, the error handler of a recipient is cached by producer, so its output is a small wrapper (`SendHelper.newSendHookProducer`) that looks up the hook on every send; that way a hook installed after the error handler was cached is still seen. **Where the hook runs, and errors.** For a send from a route the hook runs inside the error handler, so a stopped send is a real failure: `onException`, redelivery and the dead letter channel apply to it, and each redelivery runs the hook again. Per place: `to`, `toD`, `enrich`: the error handler of the route; `wireTap`: the error handler of the tapped copy; `recipientList`: the error handler of the recipient (only that recipient is redelivered); `routingSlip` and `dynamicRouter`: the error handler of the step; inside `doTry`: `doCatch`. A send from a `ProducerTemplate` has no error handler, so the caller gets the exception. The hook runs after the sending event and before the sent event, so the time taken of the sent event includes a delay of the hook. **No hook: nothing changes.** Same events in the same order, same timing (the `StopWatch` starts and stops where it did), same error handling, including the existing quirks: a producer that throws instead of calling the callback still gets no sent event at `SendProcessor`, `EventNotifierProducer` and `doInAsyncProducer`, and `ProducerTemplate.asyncSend` still fires the sending and the sent event twice (once in `doInAsyncProducer`, once in `EventNotifierCallback`; the hook runs once there). `SendHookTest#testEventsAreTheSameWithoutHook` runs nine send paths with and without a hook and compares the events. No upgrade guide entry, as nothing changes for users. **Cost without a hook.** One read of a volatile field (`getSendHook() == null`) per send, like the `isEventNotificationApplicable()` check that the same places do today. The one exception: the cached error handler of a recipient (recipient list) now calls the producer through the wrapper above, which is one more method call and the same one read per send. The hook is a field of `ExtendedCamelContext`, not a context plugin, because it is read on every send and a plugin is looked up in a map. A quick check (no JMH; a `ProducerTemplate` into a route with 10 `to` steps, and into a recipient list of 10 recipients, to a producer that does nothing; 6 alternating JVM runs of 9 x 100,000 calls each; the numbers are the average of the 6 runs) gave 1107 ns per call on main and 1130 ns with this change for the 10 `to` steps, 4282 ns and 4308 ns for the recipient list. That is within the spread between runs of the same code (4 to 8 % between the fastest and the slowest run). A variant that bui lds the recipient's error handler without the wrapper when no hook is installed was not measurably faster (4269 ns for the recipient list and 1161 ns for the 10 `to` steps, within the same spread), so the wrapper stays. **Async contract.** With a hook, every path calls the callback exactly once, and `process` returns the value the callback is called with: a stopped or skipped send, an exception from `beforeSend` (set on the exchange, no producer, no `afterSend`), an exception from `afterSend` (set on the exchange, or added as suppressed when it has failed already), a producer that throws instead of calling the callback (set on the exchange, `afterSend` runs, then it is thrown to the caller, which handles it as it does without a hook), and a delayed send (returns `false`, completes with `done(false)`). `afterSend` runs exactly once for every `beforeSend` that returned. An exchange that already has an exception when it is sent is sent as usual: only an exception the hook sets stops the send. **Delays.** A positive value from `beforeSend` does not block the thread that sends: the helper schedules the call of the producer on the scheduled thread pool that the error handler uses for asynchronous redelivery (`PluginHelper.getErrorHandlerExecutorService`), hands it to the reactive executor, as redelivery does, and the send completes asynchronously. A transacted exchange must stay on its thread, so for it the thread waits (`TimeSource.sleep`), as a redelivery of a transacted exchange does. A send from a route that waits for its delay belongs to an inflight exchange, so a graceful shutdown waits for it, as for a delayed redelivery. If the pool is shut down already, the send fails with a `RejectedExecutionException` and the producer is not called. (That pool's default rejection policy, CallerRuns, drops a task silently once the pool is shut down, so the helper checks `isShutdown()` first.) When Camel stops and a send still waits for its delay after the graceful shutdown timed out, the pool is shut down with `shutdownNow` and that send is never completed, as for a delayed asynchronous redelivery. The design document says the `delay` fault sleeps through `TimeSource`; for a send, the delay is a task on that pool instead (only a transacted exchange sleeps, through `TimeSource`). The pool comes from the `ExecutorServiceManager`, so a `camel-test-dst` world runs it in virtual time, as it runs the other timed actions of chaos. As with redelivery, a slow synchronous producer that is called after a delay runs on a thread of that pool. **Not in this PR.** The step hook in the channel (CAMEL-26338), camel-chaos itself (CAMEL-26330 and the steps after this one in the design document), and auto-detecting a `SendHook` bean from the registry. (camel-main picks up about thirty single SPIs from the registry in `DefaultConfigurationConfigurer`, such as `StartupStepRecorder` and `ReactiveExecutor`; I can add `SendHook` there in this PR if wanted. camel-chaos installs its hook itself, only when chaos is switched on.) `pollEnrich` receives, it does not send, so it has no send hook. Tests (camel-core, no sleeps; waits are latches, `MockEndpoint` waits and one Awaitility wait for the wire tap thread): - New `SendHookTest` (22 tests): the hook runs once before and once after, between the sending and the sent event, for `to`, `toD`, `enrich`, `wireTap`, `recipientList`, `routingSlip`, `multicast` (through `to`), `ProducerTemplate.send`, `ProducerTemplate.asyncSend`, a deferred producer and an asynchronous producer. A stopped send: the producer is not called, the send is redelivered twice and dead-lettered with the hook's exception, for `to`, for a recipient list (only the stopped recipient is redelivered, the other one gets one message) and for a routing slip (per step); it goes to `onException`, to `doCatch`, and back to a `ProducerTemplate` caller. A changed message reaches the producer, a changed reply reaches the next step, a skipped send does not call the producer and the route goes on, and a delayed send calls the producer from the error handler's scheduled pool, not from the caller thread. Without the hook the events are the same as with it (minus the hook calls). - New `SendHelperTest` (34 tests): the callback contract for each hook outcome (proceed, change, stop, skip, delay, `beforeSend` throws, `afterSend` throws) with a synchronous producer, an asynchronous one (completes on another thread) and one that throws, and with a transacted exchange (waits on the caller thread, and an interrupted wait fails the send with the `InterruptedException`, without calling the producer); an exchange that has failed already is sent as usual; a delay when the pool is shut down fails with `RejectedExecutionException`. A stress test: 16 threads x 500 sends with random outcomes and producers; every send completes exactly once, with the value `process` returned, and `afterSend` runs once for every `beforeSend` that returned. - New `SendHookStressTest`: 8 threads x 250 exchanges through `to`, `toD`, `recipientList` (with an asynchronous producer) and `routingSlip`, with a hook that randomly stops, skips, delays or changes each send, and a dead letter channel with redelivery: each exchange completes exactly once (one completed or failed event per exchange, every future completes), and `afterSend` runs once per `beforeSend`. - The new tests use the new SPI, so they do not compile on main. With the six places put back as on main (the SPI and `SendHelper` kept), all 23 tests of `SendHookTest` and `SendHookStressTest` fail; `SendHelperTest` tests the helper itself and passes. Three mutations of the change, run together, are each caught (7 of the first 56 tests fail): no hook around the recipient producers (`testRecipientList`, `testStoppedSendInRecipientListIsRedeliveredPerRecipient`, `testStoppedSendGoesToDoCatch`); a delayed send completing with the producer's `doneSync` instead of `false` (`SendHelperTest` DELAY with a synchronous producer, and `SendHelperTest#testStressExactlyOnce`); the hook around the whole callback of `doInAsyncProducer` instead of the producer (`testStoppedSendInRoutingSlipIsRedeliveredPerStep`: the whole slip is redelivered, `mock:ok` gets 3 messages; and `testProducerTemplateAsync`). While writing the tests, `SendHookTest` also caught a real bug in a first version: the helper's callback overrode `run()`, and as an `AsyncCallback` is a `Runnable` that the routing engine schedules, the producer was called again and again. - camel-core, the tests of the touched areas (`*EventNotifier*`, `*SendProcessor*`, `*ProducerCache*`, `*Multicast*`, `*RecipientList*`, `*RoutingSlip*`, `*WireTap*`, `*Enrich*`, `*ProducerTemplate*`, `*ErrorHandler*`, `*Redelivery*`, `*DeadLetter*`, `*DynamicRouter*`, `*SendDynamic*`, `*ToD*`, `*Sent*`, `*Sending*`, `*AsyncEndpoint*`, `*Defer*`, `*Produce*` and the new tests): 1299 tests, 4 failed: `FileProducerChecksumOfWrittenFileTest` (3) and `FilePollEnrichDynamicFileNameTest`, known failures of my local offline build without this change. - The whole camel-core suite: 8387 tests, 45 skipped; 13 failed, all unrelated to sends and known from my local offline builds without this change: `SimpleSyntaxHintsTest` (6), three OGNL hint tests of `SimpleTest`, `FileProducerChecksumOfWrittenFileTest` (3) and `ValidatorExternalResourceTest` (it reads an XSD from github.com, and the build is offline); `FilePollEnrichDynamicFileNameTest` failed twice and passed on the third run. The upstream modules ran their tests in the same build and passed (camel-util 305, camel-support 140, camel-core-model 11 and the smaller ones), except camel-xml-io and camel-yaml-io, which had 5 failures in the REST `apiDocs` model writer tests, which this change does not touch; they fail the same way on main in the same local offline build. - camel-management: 458 tests, 0 failures, run one test class at a time (`-Dcamel.surefire.parallel=false`; with `forkCount=1` and the module's parallel setting, the test classes share one MBean server and step on each other). - A second run of the same two suites, on this change and on main (the commit this branch starts from), gave the same result for every test class on both (camel-core: 8387 and 8331 tests, the same 14 failures on both, the 13 above and `FilePollEnrichDynamicFileNameTest`, which failed all three runs this time; camel-management: 458 tests, 0 failures on both); the only difference is the three new test classes. The whole camel-core suite above ran on the first version of this change; this version adds the interrupted-wait test and changes javadoc only (the new tests, `*ProducerCache*` and `*EventNotifier*`: 119 tests, 0 failures). # 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-26331) filed for the change (usually before you start working on it): [CAMEL-26331](https://issues.apache.org/jira/browse/CAMEL-26331). # 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. The new `sendHook` property is in the regenerated `ExtendedCamelContextConfigurer`. 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]
