allthingssecurity opened a new pull request, #29374: URL: https://github.com/apache/camel/pull/29374
# Description [CAMEL-26688](https://issues.apache.org/jira/browse/CAMEL-26688), part of [CAMEL-26630](https://issues.apache.org/jira/browse/CAMEL-26630) (exchange sanitizer). The first code PR of the exchange sanitizer design (#29296): step 1 of its "Order of work". It adds the step listener SPI, the Javadoc of the contracts the sanitizer will check, and the wrapper that calls the listeners. There is no sanitizer, rule or checker in this PR; nothing in Camel installs a listener yet. ## What it adds - `org.apache.camel.spi.StepListener` (camel-api, `@since 4.24`), as in the design: `onStepCall(exchange, routeId, nodeId)` returns a state for the call; `onStepReturn(state, doneSync)`, `onStepThrow(state, cause)` and `onStepCallback(state, doneSync)` (each time the callback is called, before it continues the exchange) get that state back. All methods have an empty default. A listener only observes, and many can be installed, separate from the one `StepHook` (the method names differ from `StepHook.beforeStep`, so one class can implement both). - `ExtendedCamelContext.addStepListener`, `removeStepListener` and `getStepListeners`, default methods as for `setStepHook`. `DefaultCamelContextExtension` keeps them in a copy-on-write list; the getter returns a cached unmodifiable view, and a listener equal to one already installed is ignored. - The Javadoc of the contracts, as the design states them: on `AsyncProcessor` and `AsyncCallback`, the callback is called exactly once per call, `process` does not throw (it sets the exception and calls the callback), and the `doneSync` of the callback is the value `process` returns (with one named exception, see below); on `Synchronization`, for the commit of a consumer, `onComplete` or `onFailure` runs once, where the commit is held at the end, and it moves only by a handover made by the exchange its unit of work was created for (`handoverCompletions(target)`, or `handoverCompletions()` that takes it out to be run by someone else). - The wrapper (`StepListenerProcessor`, a `DelegateAsyncProcessor` in camel-base-engine). `DefaultChannel.initChannel` adds it after the `InterceptStrategy` loop, around what the loop returns, just before `output = target`, and only when at least one listener is installed when the route is created. So the order of a step with JMX and an error handler that can redeliver is: error handler, JMX instrumentation, `StepHookProcessor`, the wrapper, the strategy wrappers, the processor. The strategies get the processor of the step as `target` and the same `next` as today; the `StepHook` is asked before the wrapper, so a step the hook fails is not a call the listeners see; `Channel.getNextProcessor()` and `Navigate` still find the processor. **With no listener installed, nothing changes:** the channel of each step reads the (empty) list once when the route is created and builds the same processor chain as today; no wrapper, no check when a step is called. The wrapper follows the listener rules of the design: the listeners of a call are read when it starts and kept for its callback (a listener removed meanwhile still gets the callback of a call it saw start); each call of a listener is in its own `try`/`catch (Throwable)`, logged at WARN once per listener (then DEBUG); the processor is called whatever `onStepCall` did; the real callback is called in a `finally`; what the processor throws is thrown on unchanged; a listener whose `onStepCall` threw gets no more events for that call. The state is held only by the wrapped callback of the call; the wrapper keeps no reference to the exchange. ## Where the design left a choice (kept minimal), or had to be read - The `doneSync` rule: Camel's own channel (`CamelInternalProcessor`) returns the value of the processor it calls, while the reactive executor runs its callback with `false`, at once or after the work the thread is running (the design notes this under R3). So the Javadoc states the rule and names this exception: a processor that hands its callback to the reactive executor, whose caller goes on through the callback. Whether every core EIP keeps the rule is for the R3 step, as the design says. - The `Synchronization` contract is written for the commit of a consumer, as the design's R2 defines it, not for every on completion: other on completions are taken out and run by someone else (aggregator, enrichers, `onCompletion`, seda), or removed with `UnitOfWork.removeSynchronization`. - `addStepListener` throws `UnsupportedOperationException` by default (as `setStepHook`); `removeStepListener` is a no-op by default and `getStepListeners` returns an empty list. - "Logged at WARN once per listener": a static, synchronized `WeakHashMap` of the listeners that already failed (weak keys, so it keeps no listener alive; listeners are compared with `equals`), touched only when a listener throws. - Whether a step gets the wrapper is decided when its route is created; a step with the wrapper reads the listeners at the start of each call, so a listener added later is called by those steps (and a step without the wrapper never calls listeners). - A processor of a step that is `ErrorHandlerAware` (multicast, recipient list, bean) gets a wrapper that is `ErrorHandlerAware` too and passes the error handler on, so `ProcessorReifier.wrapInErrorHandler` still gives it the error handler of the step, as without the wrapper. - The wrapper is built only by `DefaultChannel`, not by channels of a custom `InternalProcessorFactory`. No standby flag (an open question of the design, left out of the first version). - No upgrade guide entry: no behaviour changes, and the new `ExtendedCamelContext` methods are defaults. ## Tests New, in camel-core (deterministic: latches, no sleeps): - `StepListenerTest` (12): every step is seen; each redelivery of the error handler is a call; an `AsyncProcessor` that throws is seen as a throw and its redelivery as a call; steps without an error handler; steps of `doTry`, `doCatch` and `doFinally`, also the `doTry` step that the strategies skip; a callback on another thread (return on the caller, callback on the worker); a step that the `StepHook` fails is not a call; a failing listener does not change the routing and gets no more events for a call whose `onStepCall` threw; the exception of the processor reaches the error handler unchanged (same instance) when a listener throws from `onStepThrow`; a removed listener gets the callback of a call it saw start and nothing after; listeners in order, the same listener added twice ignored, unmodifiable view; the processor chain with a listener (error handler -> wrapper -> processor, `Navigate` unchanged). - `StepListenerNotInstalledTest` (1): with no listener, the error handler (or the channel) of every step of three routes (17 steps, with doTry/doCatch/doFinally, split, multicast, noErrorHandler and a dead letter channel) calls the processor of the step itself. - `StepListenerInterceptStrategyTest` (4): a strategy that checks the type of its `target` (as camel-opentelemetry does) still gets the `SendProcessor` of the step and `next == null`, and the listener sees the call of its chain; the caveats the design lists: with `intercept()` the listener sees `false` from the pipeline, not the processor's own `true`, and the intercept step as its own call; a strategy that returns a plain `Processor` makes the call synchronous (await manager) and hides the second callback of a processor, which the listener sees when the strategy passes the call on as an `AsyncProcessor`. New, in camel-management: - `ManagedStepListenerTest` (1): a listener together with a `StepHook` (as camel-chaos uses), JMX, the tracer and the backlog debugger: the chain order above; the hook fails the first attempt and the listener sees only the redelivered call; JMX counts 1 failed and 1 completed for the step; the tracer traces; the debugger stops at its breakpoint before the listener sees that step, and resumes. Runs (JDK 21, offline core reactor): - the 18 new tests, 20 times each: 340/340 (camel-core) and 20/20 (camel-management); - negative controls (mutants, run once): a wrapper built also without listeners fails `StepListenerNotInstalledTest`; the wrapper put before the strategy loop fails 3 tests of `StepListenerInterceptStrategyTest` (the strategy no longer gets the `SendProcessor`); `onStepCall` without its `try`/`catch` fails `testFailingListenerDoesNotChangeTheRouting`; - the related test packages with Maven (`-Dtest=org/apache/camel/{processor,impl,issues,management,builder,model,reifier}/**/*Test`): camel-core 4943 tests (1979 classes, including the `processor.intercept`, `processor.interceptor`, `processor.errorhandler` and `processor.onexception` packages and the `StepHook` tests), 0 failures, 34 skipped; camel-management 579 tests, 0 failures, 1 skipped; - the same camel-core `processor` packages and all camel-management tests once more with a pass-through listener installed in every test context (a local change to `ContextTestSupport`, not in this PR), so the wrapper is on every step, run in one JVM per module outside Maven: camel-core 3066 tests, 1 failure (a 20 ms timing tolerance in `BacklogTracerCalledRouteTimestampTest`, which passes 5/5 alone with the listener); camel-management: the one-JVM run itself fails about 90 tests that share the MBean server (96 without the listener, 84 with it), and the 3 classes that failed only with the listener pass 3/3 alone with it. The JMH benchmark the design asks for (a route with several steps, a consumer commit and an `InOnly` send to seda; no listener versus one no-op listener; the no-listener case on main and on this branch) follows in apache/camel-performance-tests, and its result is added here before this PR is merged, as the design asks. No number is claimed before that run. The R1 prototype of the design's "Validation" (step 1 there) is built on this SPI in the next PR (step 2 of the order of work); it has not been run on this SPI yet. If it shows the SPI needs a change, the change is made here before merge, so this PR can wait for it and for the JMH result. # 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): [CAMEL-26688](https://issues.apache.org/jira/browse/CAMEL-26688). # 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, camel-core and camel-management with the formatter and import-sort plugins and ran the tests above; 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]
