[jira] [Comment Edited] (CAMEL-15366) Parallel processing a poll enrich throws a RejectedExecutionException

["Alessio Villa (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/CAMEL-15366?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17170671#comment-17170671 ] 

Alessio Villa edited comment on CAMEL-15366 at 8/4/20, 9:16 AM:
----------------------------------------------------------------

Thanks Claus, your suggestion was enlightening! I've migrated a few more routes as well and it significantly sped the overall process up.

Is there any chance that something similar can be used to copy remote files too? So far, in fact, I perform this task by applying the very same approach: split a list of path/name objects and, for each corresponding file, poll enrich the message through the sftp component to download it.


was (Author: ale_amx):
Thanks Claus, your suggestion was enlightening! I've migrated a few more routes as well and it significantly sped the overall process up.

Is there any chance that something similar can be used to copy remote files too? So far, in fact, I perform this task by applying the very same approach: split a list of path/name objects and, for each corresponding file, poll enrich the message through the sftp component to download it).

> Parallel processing a poll enrich throws a RejectedExecutionException
> ---------------------------------------------------------------------
>
>                 Key: CAMEL-15366
>                 URL: https://issues.apache.org/jira/browse/CAMEL-15366
>             Project: Camel
>          Issue Type: Bug
>          Components: camel-core-engine
>    Affects Versions: 3.3.0
>         Environment: Spring Boot 2.3.0.RELEASE
> Apache Camel 3.3.0
>            Reporter: Alessio Villa
>            Priority: Major
>
> The following route receives a message containing a collection of objects representing files (path/name); it then splits the body and, for each element, reads the corresponding file (through a pollEnrich) and writes it to a different directory: 
> from("direct:copy")
>    .split(body()).streaming().parallelProcessing()
>      .pollEnrich().simple(format("file://${body.path}?fileName=RAW(${body.name})&noop=true&idempotent=false").timeout(0).end()
>      .to("file:target")
>    .end();
>  
> The source directory contains 2000 test files (txt, 1KB each) and, after around 1350-1550 files, the test fails by throwing the following exception(s):
>  
> Message History (complete message history is disabled)Message History (complete message history is disabled)---------------------------------------------------------------------------------------------------------------------------------------
> RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
> [ingestionPipelineR] [ingestionPipelineR] [                                                                              ] [    160482]
> ...
> [ingestionCopyRoute] [pollEnrich2       ] [pollEnrich[simple\\{file://${exchangeProperty.SourceDirectory}?fileName=RAW($\{bo] [         0]
>  Stacktrace
> ---------------------------------------------------------------------------------------------------------------------------------------
> org.apache.camel.CamelExchangeException: Error during poll. Exchange[]. Caused by: [java.util.concurrent.RejectedExecutionException - PollingConsumer on file://target/test-junk/source/?fileName=RAW(simple_file_515.txt)&noop=true&idempotent=false is not started, but in state: Stopped]
>     at org.apache.camel.processor.PollEnricher.process(PollEnricher.java:300) at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.doRun(RedeliveryErrorHandler.java:702)
>     at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$RedeliveryTask.run(RedeliveryErrorHandler.java:616)
>     at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:148) at org.apache.camel.impl.engine.DefaultReactiveExecutor.schedule(DefaultReactiveExecutor.java:55)
>     at org.apache.camel.processor.MulticastProcessor.lambda$schedule$1(MulticastProcessor.java:281) 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)
> Caused by: java.util.concurrent.*RejectedExecutionException*: *PollingConsumer* on [file://target/test-junk/source/?fileName=RAW(simple_file_515.txt)&noop=true&idempotent=false] *is not started, but in state: Stopped*
>     at org.apache.camel.support.EventDrivenPollingConsumer.receive(EventDrivenPollingConsumer.java:144)
>     at org.apache.camel.component.file.GenericFilePollingConsumer.receiveNoWait(GenericFilePollingConsumer.java:84)
>     at org.apache.camel.processor.PollEnricher.process(PollEnricher.java:288)
>     ... 10 common frames omitted
>  
>  
> I've also tried to use the seda component (and with the custom executor service) instead of the processingParallel EIP, without success.
>  
> Am I mistaking something? The route seems quite straightforward and I was expecting it to copy files without problems.



--
This message was sent by Atlassian Jira
(v8.3.4#803005)