flume-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From johnyru...@apache.org
Subject flume git commit: FLUME-2704. Configurable poll delay for spooling directory source
Date Sun, 17 Jan 2016 18:35:04 GMT
Repository: flume
Updated Branches:
  refs/heads/trunk e6416a070 -> af63d38fa


FLUME-2704. Configurable poll delay for spooling directory source

(Somin Mithraa via Johny Rufus)


Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/af63d38f
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/af63d38f
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/af63d38f

Branch: refs/heads/trunk
Commit: af63d38fada97a06c542ad875ef31ea3e74d53cc
Parents: e6416a0
Author: Johny Rufus <johnyrufus@apache.org>
Authored: Sun Jan 17 10:25:41 2016 -0800
Committer: Johny Rufus <johnyrufus@apache.org>
Committed: Sun Jan 17 10:25:41 2016 -0800

----------------------------------------------------------------------
 .../java/org/apache/flume/source/SpoolDirectorySource.java   | 8 ++++----
 .../source/SpoolDirectorySourceConfigurationConstants.java   | 4 ++++
 flume-ng-doc/sphinx/FlumeUserGuide.rst                       | 1 +
 3 files changed, 9 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flume/blob/af63d38f/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySource.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySource.java
b/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySource.java
index 0b11fc9..3fe947d 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySource.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySource.java
@@ -45,9 +45,6 @@ Configurable, EventDrivenSource {
   private static final Logger logger = LoggerFactory
       .getLogger(SpoolDirectorySource.class);
 
-  // Delay used when polling for new files
-  private static final int POLL_DELAY_MS = 500;
-
   /* Config options */
   private String completedSuffix;
   private String spoolDirectory;
@@ -72,6 +69,7 @@ Configurable, EventDrivenSource {
   private boolean hitChannelException = false;
   private int maxBackoff;
   private ConsumeOrder consumeOrder;
+  private int pollDelay;
 
   @Override
   public synchronized void start() {
@@ -105,7 +103,7 @@ Configurable, EventDrivenSource {
 
     Runnable runner = new SpoolDirectoryRunnable(reader, sourceCounter);
     executor.scheduleWithFixedDelay(
-        runner, 0, POLL_DELAY_MS, TimeUnit.MILLISECONDS);
+        runner, 0, pollDelay, TimeUnit.MILLISECONDS);
 
     super.start();
     logger.debug("SpoolDirectorySource source started");
@@ -168,6 +166,8 @@ Configurable, EventDrivenSource {
     consumeOrder = ConsumeOrder.valueOf(context.getString(CONSUME_ORDER, 
         DEFAULT_CONSUME_ORDER.toString()).toUpperCase(Locale.ENGLISH));
 
+    pollDelay = context.getInteger(POLL_DELAY, DEFAULT_POLL_DELAY);
+
     // "Hack" to support backwards compatibility with previous generation of
     // spooling directory source, which did not support deserializers
     Integer bufferMaxLineLength = context.getInteger(BUFFER_MAX_LINE_LENGTH);

http://git-wip-us.apache.org/repos/asf/flume/blob/af63d38f/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySourceConfigurationConstants.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySourceConfigurationConstants.java
b/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySourceConfigurationConstants.java
index 895433e..5053697 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySourceConfigurationConstants.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/source/SpoolDirectorySourceConfigurationConstants.java
@@ -93,4 +93,8 @@ public class SpoolDirectorySourceConfigurationConstants {
   }
   public static final String CONSUME_ORDER = "consumeOrder";
   public static final ConsumeOrder DEFAULT_CONSUME_ORDER = ConsumeOrder.OLDEST;    
+  
+  /** Delay(in milliseconds) used when polling for new files. The default is 500ms */
+  public static final String POLL_DELAY = "pollDelay";
+  public static final int DEFAULT_POLL_DELAY = 500;
 }

http://git-wip-us.apache.org/repos/asf/flume/blob/af63d38f/flume-ng-doc/sphinx/FlumeUserGuide.rst
----------------------------------------------------------------------
diff --git a/flume-ng-doc/sphinx/FlumeUserGuide.rst b/flume-ng-doc/sphinx/FlumeUserGuide.rst
index 897a2ca..0f8461d 100644
--- a/flume-ng-doc/sphinx/FlumeUserGuide.rst
+++ b/flume-ng-doc/sphinx/FlumeUserGuide.rst
@@ -999,6 +999,7 @@ consumeOrder          oldest          In which order files in the spooling
direc
                                       directory will be scanned to pick the oldest/youngest
file, which might be slow if there
                                       are a large number of files, while using ``random``
may cause old files to be consumed
                                       very late if new files keep coming in the spooling
directory.
+pollDelay             500             Delay (in milliseconds) used when polling for new files.
 maxBackoff            4000            The maximum time (in millis) to wait between consecutive
attempts to write to the channel(s) if the channel is full. The source will start at a low
backoff and increase it exponentially each time the channel throws a ChannelException, upto
the value specified by this parameter.
 batchSize             100             Granularity at which to batch transfer to the channel
 inputCharset          UTF-8           Character set used by deserializers that treat the
input file as text.


Mime
View raw message