[
https://issues.apache.org/jira/browse/CAMEL-15366?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Alessio Villa updated CAMEL-15366:
----------------------------------
Description:
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.
was:
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.
> 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)
