CAMEL-7278: Added owner to camel-hdfs to only include files matching owner. Thanks to Martha Obrinteschi for patch.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/239579a6 Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/239579a6 Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/239579a6 Branch: refs/heads/master Commit: 239579a603b988c28b21e9e4036de8536c0c68b1 Parents: cfb6121 Author: Claus Ibsen <davscl...@apache.org> Authored: Fri Mar 14 11:03:51 2014 +0100 Committer: Claus Ibsen <davscl...@apache.org> Committed: Fri Mar 14 11:03:51 2014 +0100 ---------------------------------------------------------------------- .../apache/camel/component/hdfs2/HdfsConfiguration.java | 9 +++++++++ .../org/apache/camel/component/hdfs2/HdfsConsumer.java | 12 ++++++++++++ 2 files changed, 21 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/239579a6/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConfiguration.java ---------------------------------------------------------------------- diff --git a/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConfiguration.java b/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConfiguration.java index e1b6093..cdb8531 100644 --- a/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConfiguration.java +++ b/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConfiguration.java @@ -52,6 +52,7 @@ public class HdfsConfiguration { private int checkIdleInterval = HdfsConstants.DEFAULT_CHECK_IDLE_INTERVAL; private List<HdfsProducer.SplitStrategy> splitStrategies; private boolean connectOnStartup = true; + private String owner; public HdfsConfiguration() { } @@ -407,4 +408,12 @@ public class HdfsConfiguration { public void setConnectOnStartup(boolean connectOnStartup) { this.connectOnStartup = connectOnStartup; } + + public String getOwner() { + return owner; + } + + public void setOwner(String owner) { + this.owner = owner; + } } http://git-wip-us.apache.org/repos/asf/camel/blob/239579a6/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConsumer.java ---------------------------------------------------------------------- diff --git a/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConsumer.java b/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConsumer.java index a437069..5b52f8b 100644 --- a/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConsumer.java +++ b/components/camel-hdfs2/src/main/java/org/apache/camel/component/hdfs2/HdfsConsumer.java @@ -120,9 +120,21 @@ public final class HdfsConsumer extends ScheduledPollConsumer { } for (FileStatus status : fileStatuses) { + if (normalFileIsDirectoryNoSuccessFile(status, info)) { continue; } + + if (config.getOwner() != null) { + // must match owner + if (!config.getOwner().equals(status.getOwner())) { + if (log.isDebugEnabled()) { + log.debug("Skipping file: {} as not matching owner: {}", status.getPath().toString(), config.getOwner()); + } + continue; + } + } + try { this.rwlock.writeLock().lock(); this.istream = HdfsInputStream.createInputStream(status.getPath().toString(), this.config);
