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/405c667b Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/405c667b Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/405c667b Branch: refs/heads/camel-2.12.x Commit: 405c667b09fe11438f7849e695c7c5e5bbcd01b1 Parents: 7567b08 Author: Claus Ibsen <davscl...@apache.org> Authored: Fri Mar 14 10:59:25 2014 +0100 Committer: Claus Ibsen <davscl...@apache.org> Committed: Fri Mar 14 11:04:21 2014 +0100 ---------------------------------------------------------------------- .../apache/camel/component/hdfs/HdfsConfiguration.java | 9 +++++++++ .../org/apache/camel/component/hdfs/HdfsConsumer.java | 12 ++++++++++++ 2 files changed, 21 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/405c667b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java ---------------------------------------------------------------------- diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java index 729fb58..13a1a06 100644 --- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConfiguration.java +++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/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/405c667b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java ---------------------------------------------------------------------- diff --git a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java index 2c78740..367f418 100644 --- a/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/HdfsConsumer.java +++ b/components/camel-hdfs/src/main/java/org/apache/camel/component/hdfs/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);
