From 7a0ed9b20094f24b35ae36a2da24d3567297cc87 Mon Sep 17 00:00:00 2001
From: Brent Halsey <brent.halsey@jivesoftware.com>
Date: Tue, 22 Feb 2011 09:20:52 -0700
Subject: [PATCH] Fix line length limitation in TailSource

---
 .../cloudera/flume/handlers/text/TailSource.java   |   60 +++++++++++++-------
 1 files changed, 40 insertions(+), 20 deletions(-)

diff --git a/src/java/com/cloudera/flume/handlers/text/TailSource.java b/src/java/com/cloudera/flume/handlers/text/TailSource.java
index 267a288..e9350f4 100644
--- a/src/java/com/cloudera/flume/handlers/text/TailSource.java
+++ b/src/java/com/cloudera/flume/handlers/text/TailSource.java
@@ -151,7 +151,8 @@ public class TailSource extends EventSource.Base {
     // For following a file name
     final File file;
     // For buffering reads
-    final ByteBuffer buf = ByteBuffer.allocateDirect(Short.MAX_VALUE);
+    int maxBufferSize = (int) Math.min(Integer.MAX_VALUE, FlumeConfiguration.get().getEventMaxSizeBytes() + 1);
+    final ByteBuffer buf = ByteBuffer.allocateDirect(maxBufferSize);
     // For closing file handles and getting FileChannels
     RandomAccessFile raf = null;
     // For reading data
@@ -309,7 +310,15 @@ public class TailSource extends EventSource.Base {
       }
     }
 
-    boolean extractLines(ByteBuffer buf) throws IOException,
+    /**
+     * Read through buffer, create events for content ending in newline.
+     * Reports whether a newline's encountered, if not and the buffer's full,
+     * that buffer is thrown away, and we skip contents up to the next newline.
+     * @param buf buffer to read
+     * @param skip whether to skip content
+     * @return whether we encountered a newline
+     */
+    boolean extractLines(ByteBuffer buf, boolean skip) throws IOException,
         InterruptedException {
       boolean madeProgress = false;
       int start = buf.position();
@@ -327,15 +336,19 @@ public class TailSource extends EventSource.Base {
           buf.mark(); // new mark.
           start = buf.position();
 
-          Event e = new EventImpl(body);
-          e.set(A_TAILSRCFILE, file.getName().getBytes());
-          sync.put(e);
+          if (!skip) {
+            Event e = new EventImpl(body);
+            e.set(A_TAILSRCFILE, file.getName().getBytes());
+            sync.put(e);
+          }
+          else {
+            skip = false;
+          }
           madeProgress = true;
         }
       }
 
-      // rewind for any left overs
-      buf.reset();
+      buf.reset();   // rewind for any left overs
       buf.compact(); // shift leftovers to front.
       return madeProgress;
     }
@@ -447,30 +460,37 @@ public class TailSource extends EventSource.Base {
       // I make this a rendezvous because this source is being pulled
       // copy data from current file pointer to EOF to dest.
       boolean madeProgress = false;
-
+      boolean skip = false;
       int rd;
       while ((rd = in.read(buf)) > 0) {
         madeProgress = true;
 
         // need char encoder to find line breaks in buf.
-        lastChannelPos += (rd < 0 ? 0 : rd); // rd == -1 if at end of
-        // stream.
+        lastChannelPos += (rd < 0 ? 0 : rd); // rd == -1 if at end of stream.
 
         int lastRd = 0;
-        boolean progress = false;
-        do {
+        boolean eventProgress = false;
 
-          if (lastRd == -1 && rd == -1) {
-            return true;
-          }
+        if (lastRd == -1 && rd == -1) {
+          return true;
+        }
 
-          buf.flip();
+        buf.flip();
 
-          // extract lines
-          extractLines(buf);
+        eventProgress = extractLines(buf, skip);
+        if (eventProgress) {
+          skip = false;
+        }
+        else {
+          if (buf.position() == buf.capacity()) {
+            // we have a full buffer w/o newlines.  We have too much data for an event, so
+            // we clear it and skip to the next newline.
+            buf.clear();
+            skip = true;
+          }
+        }
 
-          lastRd = rd;
-        } while (progress); // / potential race
+        lastRd = rd;
 
         // if the amount read catches up to the size of the file, we can fall
         // out and let another fileChannel be read. If the last buffer isn't
-- 
1.7.0.4

