[
https://issues.apache.org/jira/browse/CAMEL-15366?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17170671#comment-17170671
]
Alessio Villa commented on CAMEL-15366:
---------------------------------------
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)
