You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@camel.apache.org by "Claus Ibsen (Jira)" <ji...@apache.org> on 2022/08/11 09:48:00 UTC
[jira] [Updated] (CAMEL-18384) Split/Aggregation parallelProcessing+parallelAggregate is sequential
[ https://issues.apache.org/jira/browse/CAMEL-18384?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Claus Ibsen updated CAMEL-18384:
--------------------------------
Priority: Minor (was: Major)
> Split/Aggregation parallelProcessing+parallelAggregate is sequential
> --------------------------------------------------------------------
>
> Key: CAMEL-18384
> URL: https://issues.apache.org/jira/browse/CAMEL-18384
> Project: Camel
> Issue Type: Bug
> Components: camel-core
> Affects Versions: 3.14.0
> Reporter: Jerome Guilbaud
> Priority: Minor
>
>
> I'll need the {{aggregate}} method to be run in parallel in my split so I activated {{parallelAggregate}} option but it doesn't work as I expected it.
> {code:java}
> @RunWith(SpringJUnit4ClassRunner.class)
> public class SplitAggregationTest extends CamelTestSupport {
> private final Logger LOGGER = LoggerFactory.getLogger(SplitAggregationTest.class);
> @Override
> protected RouteBuilder createRouteBuilder() throws Exception {
> return new RouteBuilder() {
> @Override
> public void configure() throws Exception {
> from("direct:start")
> .split(body(), new AggregationStrategy() {
> @Override
> public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
> try {
> Thread.sleep(1000);
> } catch (InterruptedException e) {
> LOGGER.error("InterruptedException: ", e);
> }
> return oldExchange == null ? newExchange : oldExchange;
> }
> }).streaming().parallelProcessing().parallelAggregate()
> .log(LoggingLevel.INFO, LOGGER, "Aggreg ${body}")
> .end();
> }
> };
> }
> @Test
> public void test1() throws InterruptedException {
> Thread.sleep(5000);
> template.sendBody("direct:start", Arrays.asList("A", "B", "C", "D"));
> Thread.sleep(5000);
> } {code}
> Aggregate methods (in purple) are not run in parallel, they are waiting each other to finish before:
> !https://i.stack.imgur.com/6egxf.png!
> Associated thread dump:
> {code:java}
> "Camel (camel-2) thread #17 - Split" #36 daemon prio=5 os_prio=0 tid=0x000002317da51800 nid=0x149bc waiting on condition [0x00000062a55fe000]
> java.lang.Thread.State: WAITING (parking)
> at sun.misc.Unsafe.park(Native Method)
> - parking to wait for <0x000000076e105c70> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
> at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
> at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
> at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:870)
> at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199)
> at java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:209)
> at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:285)
> at org.apache.camel.util.concurrent.AsyncCompletionService.complete(AsyncCompletionService.java:141)
> at org.apache.camel.util.concurrent.AsyncCompletionService.access$200(AsyncCompletionService.java:30)
> at org.apache.camel.util.concurrent.AsyncCompletionService$Task.accept(AsyncCompletionService.java:168)
> at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask.lambda$null$0(MulticastProcessor.java:579)
> at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask$$Lambda$519/1049021104.done(Unknown Source)
> at org.apache.camel.AsyncCallback.run(AsyncCallback.java:44)
> at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187)
> at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:59)
> at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:348)
> at org.apache.camel.processor.MulticastProcessor$$Lambda$517/1626228981.run(Unknown Source)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> Locked ownable synchronizers:
> - <0x000000076f0ca838> (a java.util.concurrent.ThreadPoolExecutor$Worker)
> "Camel (camel-2) thread #16 - Split" #35 daemon prio=5 os_prio=0 tid=0x000002317da54000 nid=0x6a8 waiting on condition [0x00000062a54fe000]
> java.lang.Thread.State: TIMED_WAITING (sleeping)
> at java.lang.Thread.sleep(Native Method)
> at com.mybatch.camel.processor.SplitAggregationTest$1$1.aggregate(SplitAggregationTest.java:34)
> at org.apache.camel.AggregationStrategy.aggregate(AggregationStrategy.java:86)
> at org.apache.camel.processor.MulticastProcessor.doAggregateInternal(MulticastProcessor.java:894)
> at org.apache.camel.processor.MulticastProcessor.doAggregate(MulticastProcessor.java:858)
> at org.apache.camel.processor.MulticastProcessor$MulticastTask.aggregate(MulticastProcessor.java:451)
> at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask.lambda$null$0(MulticastProcessor.java:582)
> at org.apache.camel.processor.MulticastProcessor$MulticastReactiveTask$$Lambda$519/1049021104.done(Unknown Source)
> at org.apache.camel.AsyncCallback.run(AsyncCallback.java:44)
> at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187)
> at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:59)
> at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:348)
> at org.apache.camel.processor.MulticastProcessor$$Lambda$517/1626228981.run(Unknown Source)
> at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
> Locked ownable synchronizers:
> - <0x000000076e105c70> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
> - <0x000000076ef115b0> (a java.util.concurrent.ThreadPoolExecutor$Worker) {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)