You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by br...@apache.org on 2012/12/19 19:13:33 UTC
[2/2] FLUME-1632: Persist progress on each file in file spooling
client/source
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableFileInputStream.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableFileInputStream.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableFileInputStream.java
new file mode 100644
index 0000000..f9e4ec9
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableFileInputStream.java
@@ -0,0 +1,232 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flume.serialization;
+
+import com.google.common.base.Charsets;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CoderResult;
+
+/**
+ * <p/>This class makes the following assumptions:
+ * <ol>
+ * <li>The underlying file is not changing while it is being read</li>
+ * </ol>
+ *
+ * <p/>The ability to {@link #reset()} is dependent on the underlying {@link
+ * PositionTracker} instance's durability semantics.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class ResettableFileInputStream implements ResettableInputStream {
+
+ public static final int DEFAULT_BUF_SIZE = 16384;
+
+ private final File file;
+ private final PositionTracker tracker;
+ private final FileInputStream in;
+ private final FileChannel chan;
+ private final ByteBuffer buf;
+ private final CharBuffer charBuf;
+ private final byte[] byteBuf;
+ private final long fileSize;
+ private final CharsetDecoder decoder;
+ private long position;
+ private long syncPosition;
+
+ /**
+ *
+ * @param file
+ * File to read
+ *
+ * @param tracker
+ * PositionTracker implementation to make offset position durable
+ *
+ * @throws FileNotFoundException
+ */
+ public ResettableFileInputStream(File file, PositionTracker tracker)
+ throws IOException {
+ this(file, tracker, DEFAULT_BUF_SIZE, Charsets.UTF_8);
+ }
+
+ /**
+ *
+ * @param file
+ * File to read
+ *
+ * @param tracker
+ * PositionTracker implementation to make offset position durable
+ *
+ * @param bufSize
+ * Size of the underlying buffer used for input
+ *
+ * @param charset
+ * Character set used for decoding text, as necessary
+ *
+ * @throws FileNotFoundException
+ */
+ public ResettableFileInputStream(File file, PositionTracker tracker,
+ int bufSize, Charset charset)
+ throws IOException {
+ this.file = file;
+ this.tracker = tracker;
+ this.in = new FileInputStream(file);
+ this.chan = in.getChannel();
+ this.buf = ByteBuffer.allocateDirect(bufSize);
+ buf.flip();
+ this.byteBuf = new byte[1]; // single byte
+ this.charBuf = CharBuffer.allocate(1); // single char
+ charBuf.flip();
+ this.fileSize = file.length();
+ this.decoder = charset.newDecoder();
+ this.position = 0;
+ this.syncPosition = 0;
+
+ seek(tracker.getPosition());
+ }
+
+ @Override
+ public synchronized int read() throws IOException {
+ int len = read(byteBuf, 0, 1);
+ if (len == -1) {
+ return -1;
+ // len == 0 should never happen
+ } else if (len == 0) {
+ return -1;
+ } else {
+ return byteBuf[0];
+ }
+ }
+
+ @Override
+ public synchronized int read(byte[] b, int off, int len) throws IOException {
+ if (position >= fileSize) {
+ return -1;
+ }
+
+ if (!buf.hasRemaining()) {
+ refillBuf();
+ }
+
+ int rem = buf.remaining();
+ if (len > rem) {
+ len = rem;
+ }
+ buf.get(b, off, len);
+ incrPosition(len, true);
+ return len;
+ }
+
+ @Override
+ public synchronized int readChar() throws IOException {
+ if (!buf.hasRemaining()) {
+ refillBuf();
+ }
+
+ int start = buf.position();
+ charBuf.clear();
+
+ boolean isEndOfInput = false;
+ if (position >= fileSize) {
+ isEndOfInput = true;
+ }
+
+ CoderResult res = decoder.decode(buf, charBuf, isEndOfInput);
+ if (res.isMalformed() || res.isUnmappable()) {
+ res.throwException();
+ }
+
+ int delta = buf.position() - start;
+
+ charBuf.flip();
+ if (charBuf.hasRemaining()) {
+ char c = charBuf.get();
+ // don't increment the persisted location if we are in between a
+ // surrogate pair, otherwise we may never recover if we seek() to this
+ // location!
+ incrPosition(delta, !Character.isHighSurrogate(c));
+ return c;
+
+ // there may be a partial character in the decoder buffer
+ } else {
+ incrPosition(delta, false);
+ return -1;
+ }
+
+ }
+
+ private void refillBuf() throws IOException {
+ buf.compact();
+ chan.read(buf);
+ buf.flip();
+ }
+
+ @Override
+ public void mark() throws IOException {
+ tracker.storePosition(tell());
+ }
+
+ @Override
+ public void reset() throws IOException {
+ seek(tracker.getPosition());
+ }
+
+ private long tell() throws IOException {
+ return syncPosition;
+ }
+
+ private synchronized void seek(long position) throws IOException {
+ // perform underlying file seek
+ chan.position(position);
+
+ // invalidate cache
+ buf.clear();
+ buf.flip();
+
+ // clear decoder state
+ decoder.reset();
+
+ // reset position pointers
+ this.position = this.syncPosition = position;
+ }
+
+ private void incrPosition(int incr, boolean updateSyncPosition) {
+ this.position += incr;
+ if (updateSyncPosition) {
+ syncPosition = position;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ tracker.close();
+ in.close();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableInputStream.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableInputStream.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableInputStream.java
new file mode 100644
index 0000000..ae989a6
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/ResettableInputStream.java
@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * <p> This abstract class defines an interface on top of InputStream for which
+ * the stream may be <code>mark</code>ed and <code>reset</code> with no limit to
+ * the number of bytes which may have been read between the calls.
+ *
+ * <p> Any implementation of this interface guarantees that the mark position
+ * will not be invalidated by reading any number of bytes.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface ResettableInputStream extends Resettable, Closeable {
+
+ /**
+ * Read a single byte of data from the stream.
+ * @return the next byte of data, or {@code -1} if the end of the stream has
+ * been reached.
+ * @throws IOException
+ */
+ public int read() throws IOException;
+
+ /**
+ * Read multiple bytes of data from the stream.
+ * @param b the buffer into which the data is read.
+ * @param off Offset into the array {@code b} at which the data is written.
+ * @param len the maximum number of bytes to read.
+ * @return the total number of bytes read into the buffer, or {@code -1} if
+ * the end of the stream has been reached.
+ * @throws IOException
+ */
+ public int read(byte[] b, int off, int len) throws IOException;
+
+ /**
+ * <p>Read a single character.
+ *
+ * <p>Note that this may lead to returning only one character in a 2-char
+ * surrogate pair sequence. When this happens, the underlying implementation
+ * should never persist a mark between two chars of a two-char surrogate
+ * pair sequence.
+ *
+ * @return The character read, as an integer in the range 0 to 65535
+ * (0x00-0xffff), or -1 if the end of the stream has been reached
+ * @throws IOException
+ */
+ public int readChar() throws IOException;
+
+ /**
+ * Marks the current position in this input stream. A subsequent call to the
+ * <code>reset</code> method repositions this stream at the last marked
+ * position so that subsequent reads re-read the same bytes.
+ *
+ * <p> Marking a closed stream should not have any effect on the stream.
+ *
+ * @throws IOException If there is an error while setting the mark position.
+ *
+ * @see java.io.InputStream#mark(int)
+ * @see java.io.InputStream#reset()
+ */
+ @Override
+ public void mark() throws IOException;
+
+ /**
+ * Reset stream position to that set by {@link #mark()}
+ * @throws IOException
+ */
+ @Override
+ public void reset() throws IOException;
+
+ @Override
+ public void close() throws IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/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 61824d8..167193c 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
@@ -18,23 +18,23 @@
package org.apache.flume.source;
import java.io.File;
+import java.io.IOException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import org.apache.flume.Context;
-import org.apache.flume.CounterGroup;
-import org.apache.flume.Event;
-import org.apache.flume.EventDrivenSource;
-import org.apache.flume.client.avro.SpoolingFileLineReader;
+import org.apache.flume.*;
+import org.apache.flume.client.avro.ReliableSpoolingFileEventReader;
import org.apache.flume.conf.Configurable;
-import org.apache.flume.event.EventBuilder;
+import org.apache.flume.serialization.LineDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
+
+import static org.apache.flume.source
+ .SpoolDirectorySourceConfigurationConstants.*;
public class SpoolDirectorySource extends AbstractSource implements
Configurable, EventDrivenSource {
@@ -43,7 +43,7 @@ Configurable, EventDrivenSource {
.getLogger(SpoolDirectorySource.class);
// Delay used when polling for new files
- private static int POLL_DELAY_MS = 500;
+ private static final int POLL_DELAY_MS = 500;
/* Config options */
private String completedSuffix;
@@ -51,27 +51,34 @@ Configurable, EventDrivenSource {
private boolean fileHeader;
private String fileHeaderKey;
private int batchSize;
- private int bufferMaxLines;
- private int bufferMaxLineLength;
+ private String ignorePattern;
+ private File metaDirectory;
+ private String deserializerType;
+ private Context deserializerContext;
- private ScheduledExecutorService executor;
private CounterGroup counterGroup;
- private Runnable runner;
- SpoolingFileLineReader reader;
+ ReliableSpoolingFileEventReader reader;
@Override
public void start() {
- logger.info("SpoolDirectorySource source starting with directory:{}",
+ logger.info("SpoolDirectorySource source starting with directory: {}",
spoolDirectory);
- executor = Executors.newSingleThreadScheduledExecutor();
+ ScheduledExecutorService executor =
+ Executors.newSingleThreadScheduledExecutor();
counterGroup = new CounterGroup();
File directory = new File(spoolDirectory);
- reader = new SpoolingFileLineReader(directory, completedSuffix,
- bufferMaxLines, bufferMaxLineLength);
- runner = new SpoolDirectoryRunnable(reader, counterGroup);
+ try {
+ reader = new ReliableSpoolingFileEventReader(directory, completedSuffix,
+ ignorePattern, metaDirectory, fileHeader, fileHeaderKey,
+ deserializerType, deserializerContext);
+ } catch (IOException ioe) {
+ throw new FlumeException("Error instantiating spooling event parser",
+ ioe);
+ }
+ Runnable runner = new SpoolDirectoryRunnable(reader, counterGroup);
executor.scheduleWithFixedDelay(
runner, 0, POLL_DELAY_MS, TimeUnit.MILLISECONDS);
@@ -86,65 +93,70 @@ Configurable, EventDrivenSource {
@Override
public void configure(Context context) {
- spoolDirectory = context.getString(
- SpoolDirectorySourceConfigurationConstants.SPOOL_DIRECTORY);
+ spoolDirectory = context.getString(SPOOL_DIRECTORY);
Preconditions.checkState(spoolDirectory != null,
"Configuration must specify a spooling directory");
- completedSuffix = context.getString(
- SpoolDirectorySourceConfigurationConstants.SPOOLED_FILE_SUFFIX,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_SPOOLED_FILE_SUFFIX);
- fileHeader = context.getBoolean(
- SpoolDirectorySourceConfigurationConstants.FILENAME_HEADER,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_FILE_HEADER);
- fileHeaderKey = context.getString(
- SpoolDirectorySourceConfigurationConstants.FILENAME_HEADER_KEY,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_FILENAME_HEADER_KEY);
- batchSize = context.getInteger(
- SpoolDirectorySourceConfigurationConstants.BATCH_SIZE,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_BATCH_SIZE);
- bufferMaxLines = context.getInteger(
- SpoolDirectorySourceConfigurationConstants.BUFFER_MAX_LINES,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_BUFFER_MAX_LINES);
- bufferMaxLineLength = context.getInteger(
- SpoolDirectorySourceConfigurationConstants.BUFFER_MAX_LINE_LENGTH,
- SpoolDirectorySourceConfigurationConstants.DEFAULT_BUFFER_MAX_LINE_LENGTH);
- }
+ completedSuffix = context.getString(SPOOLED_FILE_SUFFIX,
+ DEFAULT_SPOOLED_FILE_SUFFIX);
+ fileHeader = context.getBoolean(FILENAME_HEADER,
+ DEFAULT_FILE_HEADER);
+ fileHeaderKey = context.getString(FILENAME_HEADER_KEY,
+ DEFAULT_FILENAME_HEADER_KEY);
+ batchSize = context.getInteger(BATCH_SIZE,
+ DEFAULT_BATCH_SIZE);
+
+ ignorePattern = context.getString(IGNORE_PAT, DFLT_IGNORE_PAT);
+ String metaDirLoc = context.getString(META_DIR, DEFAULT_META_DIR);
+
+ // if absolute path, treat as absolute
+ if (metaDirLoc.charAt(0) == '/') {
+ metaDirectory = new File(metaDirLoc);
+
+ // if relative path, treat as relative to spool directory
+ } else {
+ metaDirectory = new File(spoolDirectory, DEFAULT_META_DIR);
+ }
- private Event createEvent(String lineEntry, String filename) {
- Event out = EventBuilder.withBody(lineEntry.getBytes());
- if (fileHeader) {
- out.getHeaders().put(fileHeaderKey, filename);
+ deserializerType = context.getString(DESERIALIZER, DEFAULT_DESERIALIZER);
+ deserializerContext = new Context(context.getSubProperties(DESERIALIZER +
+ "."));
+
+ // "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);
+ if (bufferMaxLineLength != null && deserializerType != null &&
+ deserializerType.equals(DEFAULT_DESERIALIZER)) {
+ deserializerContext.put(LineDeserializer.MAXLINE_KEY,
+ bufferMaxLineLength.toString());
}
- return out;
+
}
private class SpoolDirectoryRunnable implements Runnable {
- private SpoolingFileLineReader reader;
+ private ReliableSpoolingFileEventReader reader;
private CounterGroup counterGroup;
- public SpoolDirectoryRunnable(SpoolingFileLineReader reader,
+ public SpoolDirectoryRunnable(ReliableSpoolingFileEventReader reader,
CounterGroup counterGroup) {
this.reader = reader;
this.counterGroup = counterGroup;
}
+
@Override
public void run() {
try {
while (true) {
- List<String> strings = reader.readLines(batchSize);
- if (strings.size() == 0) { break; }
- String file = reader.getLastFileRead();
- List<Event> events = Lists.newArrayList();
- for (String s: strings) {
- counterGroup.incrementAndGet("spooler.lines.read");
- events.add(createEvent(s, file));
+ List<Event> events = reader.readEvents(batchSize);
+ if (events.isEmpty()) {
+ break;
}
+ counterGroup.addAndGet("spooler.events.read", (long) events.size());
+
getChannelProcessor().processEventBatch(events);
reader.commit();
}
- }
- catch (Throwable t) {
+ } catch (Throwable t) {
logger.error("Uncaught exception in Runnable", t);
if (t instanceof Error) {
throw (Error) t;
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/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 806a661..67549e0 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
@@ -44,4 +44,16 @@ public class SpoolDirectorySourceConfigurationConstants {
/** Maximum length of line (in characters) in buffer between commits. */
public static final String BUFFER_MAX_LINE_LENGTH = "bufferMaxLineLength";
public static final int DEFAULT_BUFFER_MAX_LINE_LENGTH = 5000;
+
+ /** Pattern of files to ignore */
+ public static final String IGNORE_PAT = "ignorePattern";
+ public static final String DFLT_IGNORE_PAT = "^$"; // no effect
+
+ /** Directory to store metadata about files being processed */
+ public static final String META_DIR = "metaDir";
+ public static final String DEFAULT_META_DIR = ".flumespool";
+
+ /** Deserializer to use to parse the file data into Flume Events */
+ public static final String DESERIALIZER = "deserializer";
+ public static final String DEFAULT_DESERIALIZER = "LINE";
}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/tools/PlatformDetect.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/tools/PlatformDetect.java b/flume-ng-core/src/main/java/org/apache/flume/tools/PlatformDetect.java
new file mode 100644
index 0000000..3beec06
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/tools/PlatformDetect.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.tools;
+
+/**
+ * Utilities for platform & operating system detection
+ */
+public class PlatformDetect {
+
+ /**
+ * Detects whether we are running under Microsoft Windows.
+ * @return true if and only if we are running on a Windows system.
+ */
+ public static boolean isWindows() {
+ String os = System.getProperty("os.name");
+ boolean isWin = (os.toLowerCase().indexOf("win") >= 0);
+ return isWin;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestBufferedLineReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestBufferedLineReader.java b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestBufferedLineReader.java
index 169abe5..eb86f23 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestBufferedLineReader.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestBufferedLineReader.java
@@ -19,8 +19,11 @@
package org.apache.flume.client.avro;
+import static org.apache.flume.client.avro.TestSpoolingFileLineReader.bodyAsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.apache.flume.client.avro.TestSpoolingFileLineReader
+ .bodiesAsStrings;
import java.io.File;
import java.io.FileReader;
@@ -57,17 +60,17 @@ public class TestBufferedLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- BufferedLineReader reader = new BufferedLineReader(new FileReader(f1));
-
- assertEquals("file1line1", reader.readLine());
- assertEquals("file1line2", reader.readLine());
- assertEquals("file1line3", reader.readLine());
- assertEquals("file1line4", reader.readLine());
- assertEquals("file1line5", reader.readLine());
- assertEquals("file1line6", reader.readLine());
- assertEquals("file1line7", reader.readLine());
- assertEquals("file1line8", reader.readLine());
- assertEquals(null, reader.readLine());
+ SimpleTextLineEventReader reader = new SimpleTextLineEventReader(new FileReader(f1));
+
+ assertEquals("file1line1", bodyAsString(reader.readEvent()));
+ assertEquals("file1line2", bodyAsString(reader.readEvent()));
+ assertEquals("file1line3", bodyAsString(reader.readEvent()));
+ assertEquals("file1line4", bodyAsString(reader.readEvent()));
+ assertEquals("file1line5", bodyAsString(reader.readEvent()));
+ assertEquals("file1line6", bodyAsString(reader.readEvent()));
+ assertEquals("file1line7", bodyAsString(reader.readEvent()));
+ assertEquals("file1line8", bodyAsString(reader.readEvent()));
+ assertEquals(null, reader.readEvent());
}
@Test
@@ -77,9 +80,9 @@ public class TestBufferedLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- BufferedLineReader reader = new BufferedLineReader(new FileReader(f1));
+ SimpleTextLineEventReader reader = new SimpleTextLineEventReader(new FileReader(f1));
- List<String> out = reader.readLines(5);
+ List<String> out = bodiesAsStrings(reader.readEvents(5));
// Make sure we got every line
assertEquals(5, out.size());
@@ -97,9 +100,9 @@ public class TestBufferedLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- BufferedLineReader reader = new BufferedLineReader(new FileReader(f1));
+ SimpleTextLineEventReader reader = new SimpleTextLineEventReader(new FileReader(f1));
- List<String> out = reader.readLines(10);
+ List<String> out = bodiesAsStrings(reader.readEvents(10));
// Make sure we got exactly 8 lines
assertEquals(8, out.size());
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestSpoolingFileLineReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestSpoolingFileLineReader.java b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestSpoolingFileLineReader.java
index 740bc98..bc10243 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestSpoolingFileLineReader.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestSpoolingFileLineReader.java
@@ -19,33 +19,76 @@
package org.apache.flume.client.avro;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import com.google.common.base.Charsets;
+import com.google.common.base.Throwables;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.serialization.LineDeserializer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.io.File;
+import java.io.FileFilter;
import java.io.IOException;
import java.util.List;
-import org.apache.flume.FlumeException;
+import static org.junit.Assert.*;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
+public class TestSpoolingFileLineReader {
-import com.google.common.base.Charsets;
-import com.google.common.collect.Lists;
-import com.google.common.io.Files;
+ Logger logger = LoggerFactory.getLogger(TestSpoolingFileLineReader.class);
-public class TestSpoolingFileLineReader {
private static String completedSuffix = ".COMPLETE";
private static int bufferMaxLineLength = 500;
private static int bufferMaxLines = 30;
private File tmpDir;
+ static String bodyAsString(Event event) {
+ return new String(event.getBody());
+ }
+
+ static List<String> bodiesAsStrings(List<Event> events) {
+ List<String> bodies = Lists.newArrayListWithCapacity(events.size());
+ for (Event event : events) {
+ bodies.add(bodyAsString(event));
+ }
+ return bodies;
+ }
+
+ private ReliableSpoolingFileEventReader getParser(int maxLineLength) {
+ Context ctx = new Context();
+ ctx.put(LineDeserializer.MAXLINE_KEY, Integer.toString(maxLineLength));
+ ReliableSpoolingFileEventReader parser;
+ try {
+ parser = new ReliableSpoolingFileEventReader(tmpDir, completedSuffix, "^$",
+ new File(tmpDir, ".flumespool"), false, "^$", "LINE", ctx);
+ } catch (IOException ioe) {
+ throw Throwables.propagate(ioe);
+ }
+ return parser;
+ }
+
+ private ReliableSpoolingFileEventReader getParser() {
+ return getParser(bufferMaxLineLength);
+ }
+
+ private FileFilter directoryFilter() {
+ return new FileFilter() {
+ public boolean accept(File candidate) {
+ if (candidate.isDirectory()) {
+ return false;
+ }
+ return true;
+ }
+ };
+ }
+
@Before
public void setUp() {
tmpDir = Files.createTempDir();
@@ -54,15 +97,20 @@ public class TestSpoolingFileLineReader {
@After
public void tearDown() {
for (File f : tmpDir.listFiles()) {
+ if (f.isDirectory()) {
+ for (File sdf : f.listFiles()) {
+ sdf.delete();
+ }
+ }
f.delete();
}
tmpDir.delete();
}
@Test
- /** Create three multi-line files then read them back out. Ensures that
- * files are accessed in correct order and that lines are read correctly
- * from files. */
+ // Create three multi-line files then read them back out. Ensures that
+ // files are accessed in correct order and that lines are read correctly
+ // from files.
public void testBasicSpooling() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
@@ -72,13 +120,15 @@ public class TestSpoolingFileLineReader {
Files.write("file2line1\nfile2line2\n", f2, Charsets.UTF_8);
Files.write("file3line1\nfile3line2\n", f3, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
List<String> out = Lists.newArrayList();
for (int i = 0; i < 6; i++) {
- out.add(reader.readLine());
- reader.commit();
+ logger.info("At line {}", i);
+ String body = bodyAsString(parser.readEvent());
+ logger.debug("Seen event with body: {}", body);
+ out.add(body);
+ parser.commit();
}
// Make sure we got every line
assertTrue(out.contains("file1line1"));
@@ -88,7 +138,7 @@ public class TestSpoolingFileLineReader {
assertTrue(out.contains("file3line1"));
assertTrue(out.contains("file3line2"));
- List<File> outFiles = Lists.newArrayList(tmpDir.listFiles());
+ List<File> outFiles = Lists.newArrayList(tmpDir.listFiles(directoryFilter()));
assertEquals(3, outFiles.size());
@@ -99,20 +149,19 @@ public class TestSpoolingFileLineReader {
}
@Test
- /** Make sure this works when there are initially no files */
+ // Make sure this works when there are initially no files
public void testInitiallyEmptyDirectory() throws IOException {
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- assertNull(reader.readLine());
- assertEquals(0, reader.readLines(10).size());
+ assertNull(parser.readEvent());
+ assertEquals(0, parser.readEvents(10).size());
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
- List<String> out = reader.readLines(2);
- reader.commit();
+ List<String> out = bodiesAsStrings(parser.readEvents(2));
+ parser.commit();
// Make sure we got all of the first file
assertTrue(out.contains("file1line1"));
@@ -121,10 +170,10 @@ public class TestSpoolingFileLineReader {
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\n", f2, Charsets.UTF_8);
- reader.readLine(); // force roll
- reader.commit();
+ parser.readEvent(); // force roll
+ parser.commit();
- List<File> outFiles = Lists.newArrayList(tmpDir.listFiles());
+ List<File> outFiles = Lists.newArrayList(tmpDir.listFiles(directoryFilter()));
assertEquals(2, outFiles.size());
assertTrue(
@@ -133,19 +182,17 @@ public class TestSpoolingFileLineReader {
}
@Test(expected = IllegalStateException.class)
- /** Ensures that file immutability is enforced. */
+ // Ensures that file immutability is enforced.
public void testFileChangesDuringRead() throws IOException {
- File tmpDir1 = Files.createTempDir();
- File f1 = new File(tmpDir1.getAbsolutePath() + "/file1");
+ File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
- SpoolingFileLineReader reader1 =
- new SpoolingFileLineReader(tmpDir1, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser1 = getParser();
List<String> out = Lists.newArrayList();
- out.addAll(reader1.readLines(2));
- reader1.commit();
+ out.addAll(bodiesAsStrings(parser1.readEvents(2)));
+ parser1.commit();
assertEquals(2, out.size());
assertTrue(out.contains("file1line1"));
@@ -153,15 +200,15 @@ public class TestSpoolingFileLineReader {
Files.append("file1line3\n", f1, Charsets.UTF_8);
- out.add(reader1.readLine());
- reader1.commit();
- out.add(reader1.readLine());
- reader1.commit();
+ out.add(bodyAsString(parser1.readEvent()));
+ parser1.commit();
+ out.add(bodyAsString(parser1.readEvent()));
+ parser1.commit();
}
- /** Test when a competing destination file is found, but it matches,
- * and we are on a Windows machine. */
+ // Test when a competing destination file is found, but it matches,
+ // and we are on a Windows machine.
@Test
public void testDestinationExistsAndSameFileWindows() throws IOException {
System.setProperty("os.name", "Some version of Windows");
@@ -173,22 +220,21 @@ public class TestSpoolingFileLineReader {
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
Files.write("file1line1\nfile1line2\n", f1Completed, Charsets.UTF_8);
- SpoolingFileLineReader reader =
- new SpoolingFileLineReader(tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
List<String> out = Lists.newArrayList();
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\n", f2, Charsets.UTF_8);
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
// Make sure we got every line
@@ -199,15 +245,15 @@ public class TestSpoolingFileLineReader {
assertTrue(out.contains("file2line2"));
// Make sure original is deleted
- List<File> outFiles = Lists.newArrayList(tmpDir.listFiles());
+ List<File> outFiles = Lists.newArrayList(tmpDir.listFiles(directoryFilter()));
assertEquals(2, outFiles.size());
assertTrue(outFiles.contains(new File(tmpDir + "/file2")));
assertTrue(outFiles.contains(
new File(tmpDir + "/file1" + completedSuffix)));
}
- /** Test when a competing destination file is found, but it matches,
- * and we are not on a Windows machine. */
+ // Test when a competing destination file is found, but it matches,
+ // and we are not on a Windows machine.
@Test(expected = IllegalStateException.class)
public void testDestinationExistsAndSameFileNotOnWindows() throws IOException {
System.setProperty("os.name", "Some version of Linux");
@@ -219,29 +265,28 @@ public class TestSpoolingFileLineReader {
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
Files.write("file1line1\nfile1line2\n", f1Completed, Charsets.UTF_8);
- SpoolingFileLineReader reader =
- new SpoolingFileLineReader(tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
List<String> out = Lists.newArrayList();
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\n", f2, Charsets.UTF_8);
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
// Not reached
}
@Test
- /** Test a basic case where a commit is missed. */
+ // Test a basic case where a commit is missed.
public void testBasicCommitFailure() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
@@ -250,31 +295,30 @@ public class TestSpoolingFileLineReader {
"file1line9\nfile1line10\nfile1line11\nfile1line12\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- List<String> out1 = reader.readLines(4);
+ List<String> out1 = bodiesAsStrings(parser.readEvents(4));
assertTrue(out1.contains("file1line1"));
assertTrue(out1.contains("file1line2"));
assertTrue(out1.contains("file1line3"));
assertTrue(out1.contains("file1line4"));
- List<String> out2 = reader.readLines(4);
+ List<String> out2 = bodiesAsStrings(parser.readEvents(4));
assertTrue(out2.contains("file1line1"));
assertTrue(out2.contains("file1line2"));
assertTrue(out2.contains("file1line3"));
assertTrue(out2.contains("file1line4"));
- reader.commit();
+ parser.commit();
- List<String> out3 = reader.readLines(4);
+ List<String> out3 = bodiesAsStrings(parser.readEvents(4));
assertTrue(out3.contains("file1line5"));
assertTrue(out3.contains("file1line6"));
assertTrue(out3.contains("file1line7"));
assertTrue(out3.contains("file1line8"));
- reader.commit();
- List<String> out4 = reader.readLines(10);
+ parser.commit();
+ List<String> out4 = bodiesAsStrings(parser.readEvents(4));
assertEquals(4, out4.size());
assertTrue(out4.contains("file1line9"));
assertTrue(out4.contains("file1line10"));
@@ -283,7 +327,7 @@ public class TestSpoolingFileLineReader {
}
@Test
- /** Test a case where a commit is missed and the buffer size shrinks. */
+ // Test a case where a commit is missed and the buffer size shrinks.
public void testBasicCommitFailureAndBufferSizeChanges() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
@@ -292,45 +336,44 @@ public class TestSpoolingFileLineReader {
"file1line9\nfile1line10\nfile1line11\nfile1line12\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- List<String> out1 = reader.readLines(5);
+ List<String> out1 = bodiesAsStrings(parser.readEvents(5));
assertTrue(out1.contains("file1line1"));
assertTrue(out1.contains("file1line2"));
assertTrue(out1.contains("file1line3"));
assertTrue(out1.contains("file1line4"));
assertTrue(out1.contains("file1line5"));
- List<String> out2 = reader.readLines(2);
+ List<String> out2 = bodiesAsStrings(parser.readEvents(2));
assertTrue(out2.contains("file1line1"));
assertTrue(out2.contains("file1line2"));
- reader.commit();
- List<String> out3 = reader.readLines(2);
+ parser.commit();
+ List<String> out3 = bodiesAsStrings(parser.readEvents(2));
assertTrue(out3.contains("file1line3"));
assertTrue(out3.contains("file1line4"));
- reader.commit();
- List<String> out4 = reader.readLines(2);
+ parser.commit();
+ List<String> out4 = bodiesAsStrings(parser.readEvents(2));
assertTrue(out4.contains("file1line5"));
assertTrue(out4.contains("file1line6"));
- reader.commit();
- List<String> out5 = reader.readLines(2);
+ parser.commit();
+ List<String> out5 = bodiesAsStrings(parser.readEvents(2));
assertTrue(out5.contains("file1line7"));
assertTrue(out5.contains("file1line8"));
- reader.commit();
+ parser.commit();
- List<String> out6 = reader.readLines(15);
+ List<String> out6 = bodiesAsStrings(parser.readEvents(15));
assertTrue(out6.contains("file1line9"));
assertTrue(out6.contains("file1line10"));
assertTrue(out6.contains("file1line11"));
assertTrue(out6.contains("file1line12"));
}
- /** Test when a competing destination file is found and it does not match. */
+ // Test when a competing destination file is found and it does not match.
@Test(expected = IllegalStateException.class)
public void testDestinationExistsAndDifferentFile() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
@@ -340,45 +383,43 @@ public class TestSpoolingFileLineReader {
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
Files.write("file1line1\nfile1XXXe2\n", f1Completed, Charsets.UTF_8);
- SpoolingFileLineReader reader =
- new SpoolingFileLineReader(tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
List<String> out = Lists.newArrayList();
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\n", f2, Charsets.UTF_8);
for (int i = 0; i < 2; i++) {
- out.add(reader.readLine());
- reader.commit();
+ out.add(bodyAsString(parser.readEvent()));
+ parser.commit();
}
// Not reached
}
- /**
- * Empty files should be treated the same as other files and rolled.
- */
+ // Empty files should be treated the same as other files and rolled.
@Test
public void testBehaviorWithEmptyFile() throws IOException {
- File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
+ File f1 = new File(tmpDir.getAbsolutePath() + "/file0");
Files.touch(f1);
- SpoolingFileLineReader reader =
- new SpoolingFileLineReader(tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
+ File f2 = new File(tmpDir.getAbsolutePath() + "/file1");
Files.write("file1line1\nfile1line2\nfile1line3\nfile1line4\n" +
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f2, Charsets.UTF_8);
- List<String> out = reader.readLines(8); // Expect to skip over first file
- reader.commit();
+ // Expect to skip over first file
+ List<String> out = bodiesAsStrings(parser.readEvents(8));
+
+ parser.commit();
assertEquals(8, out.size());
assertTrue(out.contains("file1line1"));
@@ -390,15 +431,15 @@ public class TestSpoolingFileLineReader {
assertTrue(out.contains("file1line7"));
assertTrue(out.contains("file1line8"));
- assertNull(reader.readLine());
+ assertNull(parser.readEvent());
// Make sure original is deleted
- List<File> outFiles = Lists.newArrayList(tmpDir.listFiles());
+ List<File> outFiles = Lists.newArrayList(tmpDir.listFiles(directoryFilter()));
assertEquals(2, outFiles.size());
- assertTrue(outFiles.contains(
- new File(tmpDir + "/file1" + completedSuffix)));
- assertTrue(outFiles.contains(
- new File(tmpDir + "/file2" + completedSuffix)));
+ assertTrue("Outfiles should have file0 & file1: " + outFiles,
+ outFiles.contains(new File(tmpDir + "/file0" + completedSuffix)));
+ assertTrue("Outfiles should have file0 & file1: " + outFiles,
+ outFiles.contains(new File(tmpDir + "/file1" + completedSuffix)));
}
@Test
@@ -409,11 +450,10 @@ public class TestSpoolingFileLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(tmpDir,
- completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- List<String> out = reader.readLines(5);
- reader.commit();
+ List<String> out = bodiesAsStrings(parser.readEvents(5));
+ parser.commit();
// Make sure we got every line
assertEquals(5, out.size());
@@ -431,20 +471,20 @@ public class TestSpoolingFileLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(tmpDir,
- completedSuffix, bufferMaxLines, bufferMaxLineLength);
- List<String> out1 = reader.readLines(5);
- reader.commit();
+ ReliableSpoolingFileEventReader parser = getParser();
+
+ List<String> out1 = bodiesAsStrings(parser.readEvents(5));
+ parser.commit();
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\nfile2line3\nfile2line4\n" +
"file2line5\nfile2line6\nfile2line7\nfile2line8\n",
f2, Charsets.UTF_8);
- List<String> out2 = reader.readLines(5);
- reader.commit();
- List<String> out3 = reader.readLines(5);
- reader.commit();
+ List<String> out2 = bodiesAsStrings(parser.readEvents(5));
+ parser.commit();
+ List<String> out3 = bodiesAsStrings(parser.readEvents(5));
+ parser.commit();
// Should have first 5 lines of file1
assertEquals(5, out1.size());
@@ -469,7 +509,7 @@ public class TestSpoolingFileLineReader {
assertTrue(out3.contains("file2line5"));
// file1 should be moved now
- List<File> outFiles = Lists.newArrayList(tmpDir.listFiles());
+ List<File> outFiles = Lists.newArrayList(tmpDir.listFiles(directoryFilter()));
assertEquals(2, outFiles.size());
assertTrue(outFiles.contains(
new File(tmpDir + "/file1" + completedSuffix)));
@@ -477,132 +517,69 @@ public class TestSpoolingFileLineReader {
}
@Test
- /** Test the case where we read finish reading and fully commit a file, then
- * the directory is empty. */
+ // Test the case where we read finish reading and fully commit a file, then
+ // the directory is empty.
public void testEmptyDirectoryAfterCommittingFile() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
Files.write("file1line1\nfile1line2\n", f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, bufferMaxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser();
- List<String> allLines = reader.readLines(2);
+ List<String> allLines = bodiesAsStrings(parser.readEvents(2));
assertEquals(2, allLines.size());
- reader.commit();
+ parser.commit();
- List<String> empty = reader.readLines(10);
+ List<String> empty = bodiesAsStrings(parser.readEvents(10));
assertEquals(0, empty.size());
}
- @Test(expected = FlumeException.class)
- /** When a line violates the character limit, we should throw an exception,
- * even if the buffer limit is not actually exceeded. */
- public void testLineExceedsMaxLineLengthButNotBufferSize() throws IOException {
- final int maxLineLength = 15;
+ // When a line violates the character limit we should truncate it.
+ @Test
+ public void testLineExceedsMaxLineLength() throws IOException {
+ final int maxLineLength = 12;
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
Files.write("file1line1\nfile1line2\nfile1line3\nfile1line4\n" +
"file1line5\nfile1line6\nfile1line7\nfile1line8\n" +
- "reallyreallyreallyreallyreallyLongfile1line9\n" +
+ "reallyreallyreallyreallyLongLineHerefile1line9\n" +
"file1line10\nfile1line11\nfile1line12\nfile1line13\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, maxLineLength);
+ ReliableSpoolingFileEventReader parser = getParser(maxLineLength);
- List<String> out1 = reader.readLines(5);
+ List<String> out1 = bodiesAsStrings(parser.readEvents(5));
assertTrue(out1.contains("file1line1"));
assertTrue(out1.contains("file1line2"));
assertTrue(out1.contains("file1line3"));
assertTrue(out1.contains("file1line4"));
assertTrue(out1.contains("file1line5"));
- reader.commit();
-
- reader.readLines(5);
- }
-
- @Test(expected = FlumeException.class)
- /** When a line is larger than the entire buffer, we should definitely throw
- * an exception. */
- public void testLineExceedsBufferSize() throws IOException {
- final int maxLineLength = 15;
- final int bufferMaxLines = 10;
-
- // String slightly longer than buffer
- StringBuilder sb = new StringBuilder();
- for (int i = 0; i < (bufferMaxLines + 1) * maxLineLength; i++) {
- sb.append('x');
- }
-
- File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
- Files.write(sb.toString() + "\n",
- f1, Charsets.UTF_8);
-
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, maxLineLength);
-
- reader.readLines(5);
- }
-
- @Test
- /** When a line is larger than the entire buffer, we should definitely throw
- * an exception. */
- public void testCallsFailWhenReaderDisabled() throws IOException {
- final int maxLineLength = 15;
- final int bufferMaxLines = 10;
-
- // String slightly longer than buffer
- StringBuilder sb = new StringBuilder();
- for (int i = 0; i < (bufferMaxLines + 1) * maxLineLength; i++) {
- sb.append('x');
- }
-
- File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
- Files.write(sb.toString() + "\n",
- f1, Charsets.UTF_8);
+ parser.commit();
- SpoolingFileLineReader reader = new SpoolingFileLineReader(
- tmpDir, completedSuffix, bufferMaxLines, maxLineLength);
-
- boolean exceptionOnRead = false;
- boolean illegalStateOnReadLine = false;
- boolean illegalStateOnReadLines = false;
- boolean illegalStateOnCommit = false;
- try {
- reader.readLines(5);
- } catch (FlumeException e) {
- exceptionOnRead = true;
- }
- try {
- reader.readLine();
- } catch (IllegalStateException e) {
- illegalStateOnReadLine = true;
- }
- try {
- reader.readLines(5);
- } catch (IllegalStateException e) {
- illegalStateOnReadLines = true;
- }
- try {
- reader.commit();
- } catch (IllegalStateException e) {
- illegalStateOnCommit = true;
- }
-
- Assert.assertTrue("Got FlumeException when reading long line.",
- exceptionOnRead);
- Assert.assertTrue("Got IllegalStateException when reading line.",
- illegalStateOnReadLine);
- Assert.assertTrue("Got IllegalStateException when reading lines.",
- illegalStateOnReadLines);
- Assert.assertTrue("Got IllegalStateException when commiting",
- illegalStateOnCommit);
+ List<String> out2 = bodiesAsStrings(parser.readEvents(4));
+ assertTrue(out2.contains("file1line6"));
+ assertTrue(out2.contains("file1line7"));
+ assertTrue(out2.contains("file1line8"));
+ assertTrue(out2.contains("reallyreally"));
+ parser.commit();
+
+ List<String> out3 = bodiesAsStrings(parser.readEvents(5));
+ assertTrue(out3.contains("reallyreally"));
+ assertTrue(out3.contains("LongLineHere"));
+ assertTrue(out3.contains("file1line9"));
+ assertTrue(out3.contains("file1line10"));
+ assertTrue(out3.contains("file1line11"));
+ parser.commit();
+
+ List<String> out4 = bodiesAsStrings(parser.readEvents(5));
+ assertTrue(out4.contains("file1line12"));
+ assertTrue(out4.contains("file1line13"));
+ assertEquals(2, out4.size());
+ parser.commit();
+ assertEquals(0, parser.readEvents(5).size());
}
-
-
@Test
public void testNameCorrespondsToLatestRead() throws IOException {
File f1 = new File(tmpDir.getAbsolutePath() + "/file1");
@@ -610,33 +587,33 @@ public class TestSpoolingFileLineReader {
"file1line5\nfile1line6\nfile1line7\nfile1line8\n",
f1, Charsets.UTF_8);
- SpoolingFileLineReader reader = new SpoolingFileLineReader(tmpDir,
- completedSuffix, bufferMaxLines, bufferMaxLineLength);
- reader.readLines(5);
- reader.commit();
+ ReliableSpoolingFileEventReader parser = getParser();
+ parser.readEvents(5);
+ parser.commit();
- assertNotNull(reader.getLastFileRead());
- assertTrue(reader.getLastFileRead().endsWith("file1"));
+ assertNotNull(parser.getLastFileRead());
+ assertTrue(parser.getLastFileRead().endsWith("file1"));
File f2 = new File(tmpDir.getAbsolutePath() + "/file2");
Files.write("file2line1\nfile2line2\nfile2line3\nfile2line4\n" +
"file2line5\nfile2line6\nfile2line7\nfile2line8\n",
f2, Charsets.UTF_8);
- reader.readLines(5);
- reader.commit();
- assertNotNull(reader.getLastFileRead());
- assertTrue(reader.getLastFileRead().endsWith("file1"));
+ parser.readEvents(5);
+ parser.commit();
+ assertNotNull(parser.getLastFileRead());
+ assertTrue(parser.getLastFileRead().endsWith("file1"));
- reader.readLines(5);
- reader.commit();
- assertNotNull(reader.getLastFileRead());
- assertTrue(reader.getLastFileRead().endsWith("file2"));
+ parser.readEvents(5);
+ parser.commit();
+ assertNotNull(parser.getLastFileRead());
+ assertTrue(parser.getLastFileRead().endsWith("file2"));
- reader.readLines(5);
- assertTrue(reader.getLastFileRead().endsWith("file2"));
+ parser.readEvents(5);
+ assertTrue(parser.getLastFileRead().endsWith("file2"));
- reader.readLines(5);
- assertTrue(reader.getLastFileRead().endsWith("file2"));
+ parser.readEvents(5);
+ assertTrue(parser.getLastFileRead().endsWith("file2"));
}
+
}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/serialization/ResettableTestStringInputStream.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/ResettableTestStringInputStream.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/ResettableTestStringInputStream.java
new file mode 100644
index 0000000..ef8b7ba
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/ResettableTestStringInputStream.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import java.io.IOException;
+
+public class ResettableTestStringInputStream implements ResettableInputStream {
+
+ private String str;
+ int markPos = 0;
+ int curPos = 0;
+
+ /**
+ * Warning: This test class does not handle character/byte conversion at all!
+ * @param str String to use for testing
+ */
+ public ResettableTestStringInputStream(String str) {
+ this.str = str;
+ }
+
+ @Override
+ public int readChar() throws IOException {
+ if (curPos >= str.length()) {
+ return -1;
+ }
+ return str.charAt(curPos++);
+ }
+
+ @Override
+ public void mark() throws IOException {
+ markPos = curPos;
+ }
+
+ @Override
+ public void reset() throws IOException {
+ curPos = markPos;
+ }
+
+ @Override
+ public int read() throws IOException {
+ throw new UnsupportedOperationException("This test class doesn't return " +
+ "bytes!");
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ throw new UnsupportedOperationException("This test class doesn't return " +
+ "bytes!");
+ }
+
+ @Override
+ public void close() throws IOException {
+ // no-op
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
new file mode 100644
index 0000000..e52affb
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import com.google.common.base.Charsets;
+import com.google.common.io.Files;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.URL;
+
+public class TestDurablePositionTracker {
+
+ private static final Logger logger = LoggerFactory.getLogger
+ (TestDurablePositionTracker.class);
+
+ @Test
+ public void testBasicTracker() throws IOException {
+ File metaFile = File.createTempFile(getClass().getName(), ".meta");
+ metaFile.delete();
+ File dataFile = File.createTempFile(getClass().getName(), ".data");
+ Files.write("line 1\nline2\n", dataFile, Charsets.UTF_8);
+
+ final long NEW_POS = 7;
+
+ PositionTracker tracker;
+ tracker = new DurablePositionTracker(metaFile, dataFile.toString());
+ Assert.assertEquals(0, tracker.getPosition());
+ tracker.storePosition(NEW_POS);
+ Assert.assertEquals(NEW_POS, tracker.getPosition());
+ tracker.close();
+
+ // target only gets updated if the file did not exist
+ tracker = new DurablePositionTracker(metaFile, "foobar");
+ Assert.assertEquals(NEW_POS, tracker.getPosition());
+ Assert.assertEquals(dataFile.getAbsolutePath(), tracker.getTarget());
+ }
+
+ // test a valid file
+ @Test
+ public void testGoodTrackerFile() throws IOException, URISyntaxException {
+ String fileName = "/TestResettableFileInputStream_1.avro";
+ File trackerFile = new File(getClass().getResource(fileName).toURI());
+ Assert.assertTrue(trackerFile.exists());
+ PositionTracker tracker;
+ tracker = new DurablePositionTracker(trackerFile, "foo");
+ // note: 62 is the last value in this manually-created file
+ Assert.assertEquals(62, tracker.getPosition());
+ }
+
+ // test a truncated file
+ @Test
+ public void testPartialTrackerFile() throws IOException, URISyntaxException {
+ String fileName = "/TestResettableFileInputStream_1.truncated.avro";
+ File trackerFile = new File(getClass().getResource(fileName).toURI());
+ Assert.assertTrue(trackerFile.exists());
+ PositionTracker tracker;
+ tracker = new DurablePositionTracker(trackerFile, "foo");
+ // note: 25 is the last VALID value in this manually-created file
+ Assert.assertEquals(25, tracker.getPosition());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
new file mode 100644
index 0000000..1a0d561
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestLineDeserializer.java
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import com.google.common.base.Charsets;
+import junit.framework.Assert;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class TestLineDeserializer {
+
+ private String mini;
+
+ @Before
+ public void setup() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("line 1\n");
+ sb.append("line 2\n");
+ mini = sb.toString();
+ }
+
+ @Test
+ public void testSimple() throws IOException {
+ ResettableInputStream in = new ResettableTestStringInputStream(mini);
+ EventDeserializer des = new LineDeserializer(new Context(), in);
+ validateMiniParse(des);
+ }
+
+ @Test
+ public void testSimpleViaBuilder() throws IOException {
+ ResettableInputStream in = new ResettableTestStringInputStream(mini);
+ EventDeserializer.Builder builder = new LineDeserializer.Builder();
+ EventDeserializer des = builder.build(new Context(), in);
+ validateMiniParse(des);
+ }
+
+ @Test
+ public void testSimpleViaFactory() throws IOException {
+ ResettableInputStream in = new ResettableTestStringInputStream(mini);
+ EventDeserializer des;
+ des = EventDeserializerFactory.getInstance("LINE", new Context(), in);
+ validateMiniParse(des);
+ }
+
+ @Test
+ public void testBatch() throws IOException {
+ ResettableInputStream in = new ResettableTestStringInputStream(mini);
+ EventDeserializer des = new LineDeserializer(new Context(), in);
+ List<Event> events;
+
+ events = des.readEvents(1); // only try to read 1
+ Assert.assertEquals(1, events.size());
+ assertEventBodyEquals("line 1", events.get(0));
+
+ events = des.readEvents(10); // try to read more than we should have
+ Assert.assertEquals(1, events.size());
+ assertEventBodyEquals("line 2", events.get(0));
+
+ des.mark();
+ des.close();
+ }
+
+ // truncation occurs at maxLineLength boundaries
+ @Test
+ public void testMaxLineLength() throws IOException {
+ String longLine = "abcdefghijklmnopqrstuvwxyz\n";
+ Context ctx = new Context();
+ ctx.put(LineDeserializer.MAXLINE_KEY, "10");
+
+ ResettableInputStream in = new ResettableTestStringInputStream(longLine);
+ EventDeserializer des = new LineDeserializer(ctx, in);
+
+ assertEventBodyEquals("abcdefghij", des.readEvent());
+ assertEventBodyEquals("klmnopqrst", des.readEvent());
+ assertEventBodyEquals("uvwxyz", des.readEvent());
+ Assert.assertNull(des.readEvent());
+ }
+
+ /*
+ * TODO: need test for output charset
+ @Test
+ public void testOutputCharset {
+
+ }
+ */
+
+ private void assertEventBodyEquals(String expected, Event event) {
+ String bodyStr = new String(event.getBody(), Charsets.UTF_8);
+ Assert.assertEquals(expected, bodyStr);
+ }
+
+ private void validateMiniParse(EventDeserializer des) throws IOException {
+ Event evt;
+
+ evt = des.readEvent();
+ Assert.assertEquals(new String(evt.getBody()), "line 1");
+ des.mark();
+
+ evt = des.readEvent();
+ Assert.assertEquals(new String(evt.getBody()), "line 2");
+ des.reset(); // reset!
+
+ evt = des.readEvent();
+ Assert.assertEquals("Line 2 should be repeated, " +
+ "because we reset() the stream", new String(evt.getBody()), "line 2");
+
+ evt = des.readEvent();
+ Assert.assertNull("Event should be null because there are no lines " +
+ "left to read", evt);
+
+ des.mark();
+ des.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/serialization/TestReliableSpoolingFileEventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestReliableSpoolingFileEventReader.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestReliableSpoolingFileEventReader.java
new file mode 100644
index 0000000..abc1827
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestReliableSpoolingFileEventReader.java
@@ -0,0 +1,157 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import junit.framework.Assert;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.client.avro.ReliableEventReader;
+import org.apache.flume.client.avro.ReliableSpoolingFileEventReader;
+import org.apache.flume.source.SpoolDirectorySourceConfigurationConstants;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+public class TestReliableSpoolingFileEventReader {
+
+ private static final Logger logger = LoggerFactory.getLogger
+ (TestReliableSpoolingFileEventReader.class);
+
+ private static final File WORK_DIR = new File("target/test/work/" +
+ TestReliableSpoolingFileEventReader.class.getSimpleName());
+
+ @Before
+ public void setup() throws IOException, InterruptedException {
+ if (!WORK_DIR.isDirectory()) {
+ Files.createParentDirs(new File(WORK_DIR, "dummy"));
+ }
+
+ // write out a few files
+ for (int i = 0; i < 4; i++) {
+ File fileName = new File(WORK_DIR, "file"+i);
+ StringBuilder sb = new StringBuilder();
+
+ // write as many lines as the index of the file
+ for (int j = 0; j < i; j++) {
+ sb.append("file" + i + "line" + j + "\n");
+ }
+ Files.write(sb.toString(), fileName, Charsets.UTF_8);
+ }
+ Thread.sleep(1500L); // make sure timestamp is later
+ Files.write("\n", new File(WORK_DIR, "emptylineFile"), Charsets.UTF_8);
+ }
+
+ @After
+ public void tearDown() {
+
+ // delete all the files & dirs we created
+ File[] files = WORK_DIR.listFiles();
+ for (File f : files) {
+ if (f.isDirectory()) {
+ File[] subDirFiles = f.listFiles();
+ for (File sdf : subDirFiles) {
+ if (!sdf.delete()) {
+ logger.warn("Cannot delete file {}", sdf.getAbsolutePath());
+ }
+ }
+ if (!f.delete()) {
+ logger.warn("Cannot delete directory {}", f.getAbsolutePath());
+ }
+ } else {
+ if (!f.delete()) {
+ logger.warn("Cannot delete file {}", f.getAbsolutePath());
+ }
+ }
+ }
+ if (!WORK_DIR.delete()) {
+ logger.warn("Cannot delete work directory {}", WORK_DIR.getAbsolutePath());
+ }
+
+ }
+
+ // FIXME: implement ignore pattern test
+ @Ignore
+ @Test
+ public void testIgnorePattern() {
+ ReliableSpoolingFileEventReader parser;
+ }
+
+ @Test
+ public void testRepeatedCallsWithCommitAlways() throws IOException {
+ File trackerDir = new File(WORK_DIR,
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_META_DIR);
+ ReliableEventReader reader = new ReliableSpoolingFileEventReader(WORK_DIR,
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_SPOOLED_FILE_SUFFIX,
+ SpoolDirectorySourceConfigurationConstants.DFLT_IGNORE_PAT,
+ trackerDir, false, "file",
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_DESERIALIZER,
+ new Context());
+
+ final int expectedLines = 0 + 1 + 2 + 3 + 1;
+ int seenLines = 0;
+ for (int i = 0; i < 10; i++) {
+ List<Event> events = reader.readEvents(10);
+ seenLines += events.size();
+ reader.commit();
+ }
+
+ Assert.assertEquals(expectedLines, seenLines);
+ }
+
+ @Test
+ public void testRepeatedCallsWithCommitOnSuccess() throws IOException {
+ File trackerDir = new File(WORK_DIR,
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_META_DIR);
+ ReliableEventReader reader = new ReliableSpoolingFileEventReader(WORK_DIR,
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_SPOOLED_FILE_SUFFIX,
+ SpoolDirectorySourceConfigurationConstants.DFLT_IGNORE_PAT,
+ trackerDir, false, "file",
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_DESERIALIZER,
+ new Context());
+
+ final int expectedLines = 0 + 1 + 2 + 3 + 1;
+ int seenLines = 0;
+ for (int i = 0; i < 10; i++) {
+ List<Event> events = reader.readEvents(10);
+ int numEvents = events.size();
+ if (numEvents > 0) {
+ seenLines += numEvents;
+ reader.commit();
+
+ // ensure that there are files in the trackerDir
+ File[] files = trackerDir.listFiles();
+ Assert.assertNotNull(files);
+ Assert.assertTrue("Expected tracker files in tracker dir " + trackerDir
+ .getAbsolutePath(), files.length > 0);
+ }
+ }
+
+ Assert.assertEquals(expectedLines, seenLines);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
new file mode 100644
index 0000000..73e2baa
--- /dev/null
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
@@ -0,0 +1,232 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.flume.serialization;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import junit.framework.Assert;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.List;
+
+public class TestResettableFileInputStream {
+
+ private static final boolean CLEANUP = true;
+ private static final File WORK_DIR =
+ new File("target/test/work").getAbsoluteFile();
+ private static final Logger logger = LoggerFactory.getLogger
+ (TestResettableFileInputStream.class);
+
+ private File file, meta;
+
+ @Before
+ public void setup() throws Exception {
+ Files.createParentDirs(new File(WORK_DIR, "dummy"));
+ file = File.createTempFile(getClass().getSimpleName(), ".txt", WORK_DIR);
+ logger.info("Data file: {}", file);
+ meta = File.createTempFile(getClass().getSimpleName(), ".avro", WORK_DIR);
+ logger.info("PositionTracker meta file: {}", meta);
+ meta.delete(); // We want the filename but not the empty file
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ if (CLEANUP) {
+ meta.delete();
+ file.delete();
+ }
+ }
+
+ /**
+ * Ensure that we can simply read bytes from a file.
+ * @throws IOException
+ */
+ @Test
+ public void testBasicRead() throws IOException {
+ String output = singleLineFileInit(file, Charsets.UTF_8);
+
+ PositionTracker tracker = new DurablePositionTracker(meta, file.getPath());
+ ResettableInputStream in = new ResettableFileInputStream(file, tracker);
+
+ String result = readLine(in, output.length());
+ assertEquals(output, result);
+
+ String afterEOF = readLine(in, output.length());
+ assertNull(afterEOF);
+
+ in.close();
+ }
+
+ /**
+ * Ensure a reset() brings us back to the default mark (beginning of file)
+ * @throws IOException
+ */
+ @Test
+ public void testReset() throws IOException {
+ String output = singleLineFileInit(file, Charsets.UTF_8);
+
+ PositionTracker tracker = new DurablePositionTracker(meta, file.getPath());
+ ResettableInputStream in = new ResettableFileInputStream(file, tracker);
+
+ String result1 = readLine(in, output.length());
+ assertEquals(output, result1);
+
+ in.reset();
+ String result2 = readLine(in, output.length());
+ assertEquals(output, result2);
+
+ String result3 = readLine(in, output.length());
+ assertNull(result3);
+
+ in.close();
+ }
+
+ /**
+ * Ensure that marking and resetting works.
+ * @throws IOException
+ */
+ @Test
+ public void testMarkReset() throws IOException {
+ List<String> expected = multiLineFileInit(file, Charsets.UTF_8);
+
+ int MAX_LEN = 100;
+
+ PositionTracker tracker = new DurablePositionTracker(meta, file.getPath());
+ ResettableInputStream in = new ResettableFileInputStream(file, tracker);
+
+ String result0 = readLine(in, MAX_LEN);
+ assertEquals(expected.get(0), result0);
+
+ in.reset();
+
+ String result0a = readLine(in, MAX_LEN);
+ assertEquals(expected.get(0), result0a);
+
+ in.mark();
+
+ String result1 = readLine(in, MAX_LEN);
+ assertEquals(expected.get(1), result1);
+
+ in.reset();
+
+ String result1a = readLine(in, MAX_LEN);
+ assertEquals(expected.get(1), result1a);
+
+ in.mark();
+ in.close();
+ }
+
+ @Test
+ public void testResume() throws IOException {
+ List<String> expected = multiLineFileInit(file, Charsets.UTF_8);
+
+ int MAX_LEN = 100;
+
+ PositionTracker tracker = new DurablePositionTracker(meta, file.getPath());
+ ResettableInputStream in = new ResettableFileInputStream(file, tracker);
+
+ String result0 = readLine(in, MAX_LEN);
+ String result1 = readLine(in, MAX_LEN);
+ in.mark();
+
+ String result2 = readLine(in, MAX_LEN);
+ Assert.assertEquals(expected.get(2), result2);
+ String result3 = readLine(in, MAX_LEN);
+ Assert.assertEquals(expected.get(3), result3);
+ in.close();
+ tracker.close(); // redundant
+
+ // create new Tracker & RIS
+ tracker = new DurablePositionTracker(meta, file.getPath());
+ in = new ResettableFileInputStream(file, tracker);
+
+ String result2a = readLine(in, MAX_LEN);
+ String result3a = readLine(in, MAX_LEN);
+
+ Assert.assertEquals(result2, result2a);
+ Assert.assertEquals(result3, result3a);
+ }
+
+ /**
+ * Helper function to read a line from a character stream.
+ * @param in
+ * @param maxLength
+ * @return
+ * @throws IOException
+ */
+ private static String readLine(ResettableInputStream in, int maxLength)
+ throws IOException {
+
+ StringBuilder s = new StringBuilder();
+ int c;
+ int i = 1;
+ while ((c = in.readChar()) != -1) {
+ // FIXME: support \r\n
+ if (c == '\n') {
+ break;
+ }
+ //System.out.printf("seen char val: %c\n", (char)c);
+ s.append((char)c);
+
+ if (i++ > maxLength) {
+ System.out.println("Output: >" + s + "<");
+ throw new RuntimeException("Too far!");
+ }
+ }
+ if (s.length() > 0) {
+ s.append('\n');
+ return s.toString();
+ } else {
+ return null;
+ }
+ }
+
+ private static String singleLineFileInit(File file, Charset charset)
+ throws IOException {
+ String output = "This is gonna be great!\n";
+ Files.write(output.getBytes(charset), file);
+ return output;
+ }
+
+ private static List<String> multiLineFileInit(File file, Charset charset)
+ throws IOException {
+ List<String> lines = Lists.newArrayList();
+ lines.add("1. On the planet of Mars\n");
+ lines.add("2. They have clothes just like ours,\n");
+ lines.add("3. And they have the same shoes and same laces,\n");
+ lines.add("4. And they have the same charms and same graces...\n");
+ StringBuilder sb = new StringBuilder();
+ for (String line : lines) {
+ sb.append(line);
+ }
+ Files.write(sb.toString().getBytes(charset), file);
+ return lines;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/java/org/apache/flume/source/TestSpoolDirectorySource.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/source/TestSpoolDirectorySource.java b/flume-ng-core/src/test/java/org/apache/flume/source/TestSpoolDirectorySource.java
index 6e87b21..652d2a2 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/source/TestSpoolDirectorySource.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/source/TestSpoolDirectorySource.java
@@ -93,10 +93,11 @@ public class TestSpoolDirectorySource {
Transaction txn = channel.getTransaction();
txn.begin();
Event e = channel.take();
- Assert.assertNotNull(e.getHeaders());
+ Assert.assertNotNull("Event must not be null", e);
+ Assert.assertNotNull("Event headers must not be null", e.getHeaders());
Assert.assertNotNull(e.getHeaders().get("fileHeaderKeyTest"));
- Assert.assertEquals(e.getHeaders().get("fileHeaderKeyTest"),
- f1.getAbsolutePath());
+ Assert.assertEquals(f1.getAbsolutePath(),
+ e.getHeaders().get("fileHeaderKeyTest"));
txn.commit();
txn.close();
}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.avro
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.avro b/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.avro
new file mode 100644
index 0000000..004d463
Binary files /dev/null and b/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.avro differ
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.truncated.avro
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.truncated.avro b/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.truncated.avro
new file mode 100644
index 0000000..31b217b
Binary files /dev/null and b/flume-ng-core/src/test/resources/TestResettableFileInputStream_1.truncated.avro differ
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index b934c1d..c27a980 100644
--- a/pom.xml
+++ b/pom.xml
@@ -515,8 +515,8 @@ limitations under the License.
<version>0.7</version>
<executions>
<execution>
- <id>test.rat</id>
- <phase>test</phase>
+ <id>verify.rat</id>
+ <phase>verify</phase>
<goals>
<goal>check</goal>
</goals>
@@ -532,6 +532,7 @@ limitations under the License.
<exclude>**/*.diff</exclude>
<exclude>**/*.patch</exclude>
<exclude>**/*.avsc</exclude>
+ <exclude>**/*.avro</exclude>
<exclude>**/docs/**</exclude>
</excludes>
</configuration>