Sanitizer

Each time the processor of a step is called (each attempt, also each redelivery of the error handler), its callback must be called exactly once, and process(exchange, callback) must not throw. A step is any node of a route with its node id (to, process, split, …​). It reports:

An exception that a processor throws on through an outer step is reported once, at the step that threw it first.

A plain Processor does not break the rule when it throws: the adapter Camel puts around it (DelegateSyncProcessor for process(…​), the bridge of AsyncProcessorConverterHelper, and the bean component for a bean that is a Processor) sets what it throws on the exchange, also an Error such as an AssertionError of a test, and then calls the callback. The exchange fails, and nothing is reported. A VirtualMachineError (such as an OutOfMemoryError or a StackOverflowError) is the exception: the adapter does not catch it, so it is thrown on after the callback, as in earlier versions, and is reported as thrown after the callback (THREW_AFTER_CALLBACK).

A report looks like this:

The processor named is the one of the step, looking through the delegates Camel puts around a processor (such as the one that process(…​) uses for an AsyncProcessor that is not a Service); for bean(…​) it is the class of the bean, not Camel’s BeanProcessor. For a step that sends to an endpoint, such as to, that is Camel’s SendProcessor, and the report adds the endpoint (its URI, sanitized, and its class), as the producer that broke the rule is behind it. The stacks start at the code that called the callback, or at what threw, so they show the class of the producer or the bean. The report holds ids, names, threads, stacks, and the class and message of an exception; never bodies or headers. An exception message can carry message data, so the report is for development.

The check is at the processor of each step, below the error handler of the step, as the step listener SPI of camel-core sees it (see StepListener). A processor that is not a step, such as the producer of a recipient of a recipient list, is seen through its step. An InterceptStrategy can hide part of the call: intercept() calls the processor later, so its own return or throw is not seen, and a strategy that returns a plain Processor makes a second callback of an asynchronous processor go to the await manager, not to the sanitizer.

A call whose callback has not come is not a violation, as a slow endpoint looks the same, but it is what a hung exchange shows. When nothing happened in an exchange for camel.sanitizer.pendingTimeout (default 60s: no call of a step, no return, throw or callback, also in its parts), the sanitizer logs a WARN with its innermost pending call: the step whose callback the exchange waits for, whether process returned, threw or has not returned yet (then with the stack of the thread that is still in it), and the enclosing calls that wait for it. The parts of an exchange are the correlated copies that a split, multicast, recipient list, enrich, kamelet, circuit breaker or a to("kamelet:…​") step of it waits for: a copy that carries the id of the exchange as its correlation id while such a step is the innermost pending call of the exchange. Other copies, such as the one a seda producer queues, are exchanges of their own (unless the producing exchange is already in such a step when the copy starts). So a long split, or a to("direct:…​") whose route is still busy, is not reported, and a step inside them that does not call back is. An exchange is not reported while it waits for its parts and they still have pending calls. Each call is reported once; if its callback comes later, an INFO line says after how long. A call that threw stays pending until its callback comes or the exchange is done (Camel catches the throw and fails the exchange; on older Camel versions the exchange hung there). The parts of a split with shareUnitOfWork() have no done event of their own: what they threw goes when their exchange is done. A call of a part that has not called back stays when its exchange is done, and is reported: the part may hang after the step that waited for it gave up, such as a split with a timeout. A recipient of a recipient list is not a step: a producer of a recipient that does not call back shows as the recipient list step still pending. Steps that wait on purpose for longer than the timeout are reported too: a delay or a throttle, a redelivery delay, a pollEnrich that waits, a processor that waits for something, or one that sends with a ProducerTemplate and waits for the reply. So are steps that wait for an exchange copy of their own, as the consumer side works on that copy, not on a part: a request-reply to over seda or vm whose consumer is slow but still making progress, and an aggregate step whose completion runs the aggregated exchange synchronously in the caller. A chaos delay on a step (a step= rule of camel-chaos), or at the entry of a route (a route= rule), runs before the step is called, so a delay longer than pendingTimeout shows up at the enclosing step that waits (such as a split, or the to of the calling route), and not at all at the top level of a route. A delay of an endpoint= or component= rule at a send (at=send) runs inside the send, so it shows at that to step. At a poll (at=poll) there is no exchange yet, so it is not seen; at the send of a dead letter channel (at=dlq) the error handler runs it after the failed step called back, so it shows like a step= delay.

The warnings never fail a test, and are not charged to it. When the context stops, the calls still pending then are listed at WARN (with reportOnShutdown), and the report on shutdown counts the warnings. The pending calls are kept in a table that holds ids and names, not the exchanges, with a bound of 10000 calls. The table is split in stripes, each with its own lock (twice the number of processors, rounded up to a power of two, at most 32), so routing threads that work on unrelated exchanges do not wait for each other; an exchange, its parts and the copies made from it while it is in the table are in one stripe. Each stripe has a share of the bound (10000 divided by the number of stripes), and the share is soft: a stripe past it drops its oldest calls only while all the stripes together hold more than 10000, so an asynchronous parallel split with many parts in flight is still watched. When calls are dropped, one WARN says that the checks are incomplete, and the report counts the calls dropped. The timeout is measured with System.nanoTime(), not with the time source of camel-test-dst, so a virtual clock does not stretch it.

To know when an exchange is done, the sanitizer adds an event notifier of the exchanges completed and failed (one notifier, shared with R2, which needs it too), so with the sanitizer on the exchange events of the context are fired (and the notifier is listed with the others, also in JMX). A pending timeout of 0 turns the warning off, and with it the table and the list when the context stops (and the notifier, unless R2 is checked). The warning follows the callbacks of R1, so with camel.sanitizer.checks=commit only it is off too.

The parts of a split, multicast or recipient list are in the stripe of their exchange. With parallelProcessing(), the threads that run the parts therefore take the same lock of the table at each step call, return and callback. The lock is held briefly, but it costs more when the parts run on several threads, and it orders those threads, which can hide a race between parts that is not synchronized otherwise. To hunt races between the parts of a parallel split, set camel.sanitizer.pendingTimeout=0: the other checks of R1 stay on, and the table and its lock are not used. R2 takes one lock of its records of the context when an exchange or a part gets its unit of work, at each add, handover and run of an on completion, and when an exchange is done; in a test, where R2 is on by default, set camel.sanitizer.checks=callback as well to leave that out too.

R2 is checked by default only in a test (a CamelContext built on a thread the JUnit extension marked); elsewhere, such as with camel run or profile dev, camel.sanitizer.checks=callback,commit turns it on.

A commit is the on completion (Synchronization) that a consumer adds to the exchange before it routes it (exchange.getExchangeExtension().addOnCompletion(…​)), to commit or roll back its message when the exchange is done, such as the file consumer moving or deleting the file. In the sanitizer, the commits of an exchange are the on completions it holds when it gets the unit of work created for it; on completions added later (such as by the tracer, or a route onCompletion) are not commits. A commit can move: it is handed over to a copy that is routed on its own (such as an InOnly exchange sent to seda), and then runs when the copy is done, with the outcome of the copy. The rule, as the Javadoc of Synchronization states it: each add of a commit runs onComplete or onFailure exactly once, where it is held at the end, and while it is in a unit of work it moves only by a handover of the exchange that unit of work was created for. It reports:

A report of R2 looks like this (a split with shareUnitOfWork before CAMEL-26576, where a part sent to seda took the commit of the file):

The exchange, route, consumer and endpoint are the ones the commit was added for; the step is the one that ran on the thread where the violation was found. The commit is named by the class of its on completion (for @ExpectSanitizerViolation and SanitizerViolation.getProcessor()).

R2 follows each add of a commit, by the identity of its on completion, as the on completion listener SPI of camel-core reports them (see OnCompletionListener): one on completion can be added to many exchanges. On completions that are taken out of an exchange as a list to be run by someone else (handoverCompletions(), as the seda consumer does with multipleConsumers) are checked for running once only; nothing tells which exchange runs them, so they are matched by their on completion alone, and one on completion object shared by many exchanges can then be matched to another exchange’s record. A commit in a UnitOfWork of a user that does not tell what it hands over, or does not report its adds, is not reported as never run. A consumer that commits in its own code after the callback (such as JMS with a transacted session) is covered by R1 only. The records keep no exchange; the commits that wait in exchanges without a unit of work (such as copies in a seda queue) and the commits taken out that have not run yet are bounded (10 000): past it the oldest commit taken out is no longer followed first, then the oldest waiting in an exchange; one WARN says so and the report counts them. A commit taken out that ran is kept (the last 1000) to see a second run.

A commit that never ran because the context stopped while it waited in a copy that was never routed is found when the context stops. It is charged like any violation: to the test that runs then (a context stopped at the end of each test), or to the test class (a context stopped after its last test); a test that stops a context with such a copy on purpose says so with @ExpectSanitizerViolation(Violation.COMMIT_NEVER_RUN) and stops it in the test.

To find the commits that never ran, R2 uses the event notifier of the exchanges completed and failed that the still pending warning uses too (one notifier for both, added when either is on).

With camel.sanitizer.checks=commit only, the step listener is still installed (each step is still wrapped, and the report still counts the calls of steps), so a report of R2 can name the step that handed a commit over.