Camel split then aggregate exception propagation

Viewed 29

My project need a split then aggregate operation rather than the integrated split+aggregate but I don't have any clue about how propagate an exception to stop further processing.

In the example below I'll need to be able to not make the last log if an exception was thrown in post aggregation.

@RunWith(SpringJUnit4ClassRunner.class)
public class AggregationExceptionTest extends CamelTestSupport {
private final Logger LOGGER = LoggerFactory.getLogger(AggregationExceptionTest.class);

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start")    
                    .split(body()).streaming().stopOnException().parallelProcessing()
                        .aggregate(new AggregationStrategy() {
                            @Override
                            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                                if (newExchange.getException() != null) {
                                    LOGGER.info("exception propagated");
                                }
                               return oldExchange==null?newExchange:oldExchange;
                            }
                        }).constant(true)
                        .completionSize(1).completionTimeout(500)
                            .log(LoggingLevel.INFO, LOGGER, "Aggreg ${body}")
                            .throwException(Exception.class, "propagate plz")
                        .end()
                    .end()
                    .process(e -> {
                        LOGGER.info("I don't want to be seen, because of {}", e.getException());
                    });
            }
        };
    }
    
    @Test
    public void test1() throws InterruptedException {
        template.sendBody("direct:start", Arrays.asList("A", "B", "C", "D"));
        
        Thread.sleep(5000);
    }
}

The exception thrown is never visible in the aggregate method.

1 Answers

I'm not sure this is possible. Aggregator output is completely separated from the route that invokes it. It also operates in its own thread, unless you supply it a SynchronousExecutorService instance.

You might try the SynchronousExecutorService option:

.aggregate(constant(1), new YourAggregator())
    .executorService(new SynchronousExecutorService())

Before your split, set a Map instance in the header with a key called exception, value null initially. Then in the aggregator output, if an exception occurs, update that Map key with the exception. Then in your split, you can check the Map key for an Exception instance, and do whatever you want, e.g., splitter's stopOnException option.

Related