svn commit: r1309627 - /incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java

[pr...@apache.org]

Author: prasadm
Date: Wed Apr  4 23:51:43 2012
New Revision: 1309627

URL: http://svn.apache.org/viewvc?rev=1309627&view=rev
Log:
FLUME-1102: HDFSEventSink rollInterval is broken
(Mike Percy via Prasad Mujumdar)
.

Modified:
    incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java

Modified: incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java
URL: http://svn.apache.org/viewvc/incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java?rev=1309627&r1=1309626&r2=1309627&view=diff
==============================================================================
--- incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java (original)
+++ incubator/flume/trunk/flume-ng-sinks/flume-hdfs-sink/src/main/java/org/apache/flume/sink/hdfs/BucketWriter.java Wed Apr  4 23:51:43 2012
@@ -48,7 +48,7 @@ public class BucketWriter {
   private FlumeFormatter formatter;
   private long eventCounter;
   private long processSize;
-  private long lastProcessTime;
+  private long lastRollTime;
   private long batchCounter;
   private String filePath;
   private long rollInterval;
@@ -63,7 +63,7 @@ public class BucketWriter {
   private void resetCounters() {
     eventCounter = 0;
     processSize = 0;
-    lastProcessTime = 0;
+    lastRollTime = System.currentTimeMillis();
     batchCounter = 0;
   }
 
@@ -149,7 +149,6 @@ public class BucketWriter {
 
     // update statistics
     processSize += e.getBody().length;
-    lastProcessTime = System.currentTimeMillis();
     eventCounter++;
     batchCounter++;
 
@@ -164,18 +163,22 @@ public class BucketWriter {
   }
 
   // check if time to rotate the file
-  public boolean shouldRotate() {
+  private boolean shouldRotate() {
     boolean doRotate = false;
 
-    if ((rollInterval > 0)
-        && (rollInterval < (System.currentTimeMillis() - lastProcessTime) / 1000))
+    long elapsed = (System.currentTimeMillis() - lastRollTime) / 1000L;
+    if ((rollInterval > 0) && (rollInterval <= elapsed)) {
+      LOG.debug("rolling: rollTime: {}, elapsed: {}", rollInterval, elapsed);
       doRotate = true;
+    }
+
     if ((rollCount > 0) && (rollCount <= eventCounter)) {
-      eventCounter = 0;
+      LOG.debug("rolling: rollCount: {}, events: {}", rollCount, eventCounter);
       doRotate = true;
     }
+
     if ((rollSize > 0) && (rollSize <= processSize)) {
-      processSize = 0;
+      LOG.debug("rolling: rollSize: {}, bytes: {}", rollSize, processSize);
       doRotate = true;
     }