Details
-
Bug
-
Status: Resolved
-
Major
-
Resolution: Not A Problem
-
2.12.2
-
None
-
Unknown
Description
Processors attached to aggregators can not throw exceptions, they will be swallowed. This makes it impossible to propagate exceptions from camel to, for instance, a CXF endpoint or custom JAX-WS service.
This route and test illustrates the problem:
public class AggregatorBug extends RouteBuilder { @Override public void configure() throws Exception { from("direct:start") .routeId(getClass().getSimpleName()) .aggregate().header("cheese") .aggregationStrategy(new GroupedExchangeAggregationStrategy()) .completionSize(1) .process(new Processor() { @Override public void process(Exchange exchange) throws Exception { throw new RuntimeException(); } }); } }
public class AggregatorBugTest extends CamelTestSupport { @EndpointInject(uri = "mock:beforeException") protected MockEndpoint beforeException; @Override public boolean isUseAdviceWith() { return true; } @Override protected RouteBuilder createRouteBuilder() throws Exception { return new AggregatorBug(); } @Override protected void doPostSetup() throws Exception { context.getRouteDefinition(AggregatorBug.class.getSimpleName()).adviceWith(context, new AdviceWithRouteBuilder() { @Override public void configure() throws Exception { weaveByType(ProcessDefinition.class).before().to("mock:beforeException"); } }); context.start(); } @Test public void exceptions_are_swallowed() throws Exception { try { template.sendBodyAndHeader("direct:start", 42, "cheese", "foo"); fail("Expected exception"); } catch (CamelExecutionException e) { beforeException.expectedMessageCount(1); } } }
From what I can see, the problem is located in AggregatorProcessor (github link). The output from doAggregate is ignored, even though it contains the exchange with the exception. The exchange that is used from here on out is the original input to the route, not the aggregated output.
The only way I can see that would trigger the exception would be to add a custom errorHandler that re-throwed the exception. Unfortunately this is also impossible. The errorHandler on AggregatorProcessor is set in the constructor, and there is no API to set the errorHandler from AggregatorDefinition. The errorHandler set on the route is also ignored by AggregatorProcessor (it will though be invoked with the original not-aggregated exchange).
This problem has been reported earlier in CAMEL-1546. The proposed solution with using a seda route is not satisfactory for cxf endpoints since the calling thread would return.