[
https://issues.apache.org/jira/browse/CAMEL-7318?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14348469#comment-14348469
]
Grzegorz Grzybek commented on CAMEL-7318:
-----------------------------------------
Actually the file is not being read twice.
I've setup two consumers and watched the behavior under debugger.
This code you've mentioned (in
org.apache.camel.component.hdfs2.HdfsInputStream#createInputStream()):
{code:java}
info.getFileSystem().rename(new Path(ret.actualPath), new
Path(ret.suffixedPath));
{code}
doesn't return {{false}} if can't rename, it throws exception with the
following stack trace ("file://" case):
{noformat}
java.io.FileNotFoundException: File
file:/data/ggrzybek/sources/github.com/grgrzybek/camel/components/camel-hdfs2/target/test/multiple-consumers/file-0194.txt
does not exist
at
org.apache.hadoop.fs.RawLocalFileSystem.deprecatedGetFileStatus(RawLocalFileSystem.java:511)
at
org.apache.hadoop.fs.RawLocalFileSystem.getFileLinkStatusInternal(RawLocalFileSystem.java:722)
at
org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:501)
at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:337)
at org.apache.hadoop.fs.FileUtil.copy(FileUtil.java:289)
at
org.apache.hadoop.fs.RawLocalFileSystem.rename(RawLocalFileSystem.java:334)
at
org.apache.hadoop.fs.ChecksumFileSystem.rename(ChecksumFileSystem.java:503)
at
org.apache.camel.component.hdfs2.HdfsInputStream.createInputStream(HdfsInputStream.java:49)
at
org.apache.camel.component.hdfs2.HdfsConsumer.doPoll(HdfsConsumer.java:140)
at
org.apache.camel.component.hdfs2.HdfsConsumer.poll(HdfsConsumer.java:98)
at
org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:174)
at
org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:101)
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
{noformat}
so then Camel invokes
org.apache.camel.spi.PollingConsumerPollStrategy#rollback() and simply skips
this file and move to the next one.
I'll check how it works in "hdfs://" case.
> Concurrency on HDFS Consumer not working efficiently
> ----------------------------------------------------
>
> Key: CAMEL-7318
> URL: https://issues.apache.org/jira/browse/CAMEL-7318
> Project: Camel
> Issue Type: Bug
> Components: camel-hdfs
> Affects Versions: 2.11.2
> Reporter: Martha Obrinteschi
> Assignee: Grzegorz Grzybek
> Priority: Minor
> Labels: concurrency, parallel
>
> If we have two HDFS consumers the files are being processed twice (once by
> each consumer, waiting one after the other) so the consumers are not working
> in parallel. If we add this the consumers will work as a team and the
> transfer will go faster.
> This happens because there is no exception thrown (the rename method just
> returns true or false and everything carries on as nothing would have
> happened :).
> In order to fix this we could add in the HdfsInputStream: 49
> boolean tf = info.getFileSystem().rename(new Path(ret.actualPath), new
> Path(ret.suffixedPath));
> if (!tf) {
> throw new IOException("This exception is thrown because the rename did
> not succeded.");
> }
> And also in the HdfsConsumer: 150
> try {
> this.rwlock.writeLock().lock();
> this.istream =
> HdfsInputStream.createInputStream(fileStatuses[i].getPath().toString(),
> this.config);
> } catch (IOException ioe) {
> log.info(ioe.getMessage() + " If the rename fails we move on to
> the next file.");
> continue;
> }
>
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
