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
[1/2] git commit: FLUME-1632: Persist progress on each file in file
spooling client/source
Updated Branches:
refs/heads/trunk bd981bddd -> fad7251db
FLUME-1632: Persist progress on each file in file spooling client/source
(Mike Percy via Brock Noland)
Project: http://git-wip-us.apache.org/repos/asf/flume/repo
Commit: http://git-wip-us.apache.org/repos/asf/flume/commit/fad7251d
Tree: http://git-wip-us.apache.org/repos/asf/flume/tree/fad7251d
Diff: http://git-wip-us.apache.org/repos/asf/flume/diff/fad7251d
Branch: refs/heads/trunk
Commit: fad7251dbc13e74bb070691e7a5b643f2402b901
Parents: bd981bd
Author: Brock Noland <br...@apache.org>
Authored: Wed Dec 19 12:12:51 2012 -0600
Committer: Brock Noland <br...@apache.org>
Committed: Wed Dec 19 12:12:51 2012 -0600
----------------------------------------------------------------------
flume-ng-core/pom.xml | 32 +
.../src/main/avro/TransferStateFileMeta.avsc | 8 +
.../apache/flume/client/avro/AvroCLIClient.java | 50 +-
.../flume/client/avro/BufferedLineReader.java | 67 ---
.../org/apache/flume/client/avro/EventReader.java | 58 ++
.../org/apache/flume/client/avro/LineReader.java | 51 --
.../flume/client/avro/ReliableEventReader.java | 42 ++
.../avro/ReliableSpoolingFileEventReader.java | 415 ++++++++++++++
.../client/avro/SimpleTextLineEventReader.java | 76 +++
.../flume/client/avro/SpoolingFileLineReader.java | 361 ------------
.../serialization/DurablePositionTracker.java | 202 +++++++
.../flume/serialization/EventDeserializer.java | 94 +++
.../serialization/EventDeserializerFactory.java | 88 +++
.../flume/serialization/EventDeserializerType.java | 40 ++
.../org/apache/flume/serialization/EventSerDe.java | 49 ++
.../flume/serialization/EventSerializer.java | 2 +-
.../serialization/EventSerializerFactory.java | 8 +-
.../flume/serialization/EventSerializerType.java | 5 +
.../flume/serialization/LineDeserializer.java | 167 ++++++
.../flume/serialization/PositionTracker.java | 36 ++
.../org/apache/flume/serialization/Resettable.java | 47 ++
.../serialization/ResettableFileInputStream.java | 232 ++++++++
.../flume/serialization/ResettableInputStream.java | 96 ++++
.../apache/flume/source/SpoolDirectorySource.java | 122 +++--
...SpoolDirectorySourceConfigurationConstants.java | 12 +
.../org/apache/flume/tools/PlatformDetect.java | 35 ++
.../flume/client/avro/TestBufferedLineReader.java | 33 +-
.../client/avro/TestSpoolingFileLineReader.java | 441 +++++++--------
.../ResettableTestStringInputStream.java | 70 +++
.../serialization/TestDurablePositionTracker.java | 83 +++
.../flume/serialization/TestLineDeserializer.java | 134 +++++
.../TestReliableSpoolingFileEventReader.java | 157 +++++
.../TestResettableFileInputStream.java | 232 ++++++++
.../flume/source/TestSpoolDirectorySource.java | 7 +-
.../resources/TestResettableFileInputStream_1.avro | Bin 0 -> 322 bytes
.../TestResettableFileInputStream_1.truncated.avro | Bin 0 -> 318 bytes
pom.xml | 5 +-
37 files changed, 2741 insertions(+), 816 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/pom.xml
----------------------------------------------------------------------
diff --git a/flume-ng-core/pom.xml b/flume-ng-core/pom.xml
index 0224519..ba414bc 100644
--- a/flume-ng-core/pom.xml
+++ b/flume-ng-core/pom.xml
@@ -33,11 +33,43 @@ limitations under the License.
<plugins>
<plugin>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>schema</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>com.thoughtworks.paranamer</groupId>
+ <artifactId>paranamer-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>run</id>
+ <configuration>
+ <sourceDirectory>${project.build.directory}/generated-sources/avro</sourceDirectory>
+ <outputDirectory>${project.build.directory}/classes</outputDirectory>
+ </configuration>
+ <goals>
+ <goal>generate</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
</plugin>
+
</plugins>
</build>
+
<profiles>
<profile>
<id>not-windows</id>
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/avro/TransferStateFileMeta.avsc
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/avro/TransferStateFileMeta.avsc b/flume-ng-core/src/main/avro/TransferStateFileMeta.avsc
new file mode 100644
index 0000000..f290231
--- /dev/null
+++ b/flume-ng-core/src/main/avro/TransferStateFileMeta.avsc
@@ -0,0 +1,8 @@
+{
+ "namespace": "org.apache.flume.serialization",
+ "type": "record",
+ "name": "TransferStateFileMeta",
+ "fields": [
+ {"name": "offset", "type": "long"}
+ ]
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/AvroCLIClient.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/AvroCLIClient.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/AvroCLIClient.java
index 37e9ffa..f00ce24 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/client/avro/AvroCLIClient.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/client/avro/AvroCLIClient.java
@@ -24,7 +24,6 @@ import java.io.FileInputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
-import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -36,18 +35,19 @@ import org.apache.commons.cli.GnuParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
+import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.FlumeException;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
-import org.apache.flume.event.EventBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Charsets;
-import com.google.common.collect.Lists;
-
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
public class AvroCLIClient {
private static final Logger logger = LoggerFactory
@@ -73,8 +73,7 @@ public class AvroCLIClient {
} catch (ParseException e) {
logger.error("Unable to parse command line options - {}", e.getMessage());
} catch (IOException e) {
- logger.error("Unable to send data to Flume - {}", e.getMessage());
- logger.debug("Exception follows.", e);
+ logger.error("Unable to send data to Flume. Exception follows.", e);
} catch (FlumeException e) {
logger.error("Unable to open connection to Flume. Exception follows.", e);
} catch (EventDeliveryException e) {
@@ -168,40 +167,34 @@ public class AvroCLIClient {
return true;
}
- private void setHeaders(Event event) {
- event.setHeaders(headers);
- }
-
private void run() throws IOException, FlumeException,
EventDeliveryException {
- LineReader reader = null;
+ EventReader reader = null;
RpcClient rpcClient = RpcClientFactory.getDefaultInstance(hostname, port,
BATCH_SIZE);
try {
if (fileName != null) {
- reader = new BufferedLineReader(new FileReader(new File(fileName)));
+ reader = new SimpleTextLineEventReader(new FileReader(new File(fileName)));
} else if (dirName != null) {
- reader = new SpoolingFileLineReader(new File(dirName), ".COMPLETED",
- BATCH_SIZE, MAX_LINE_LENGTH);
+ reader = new ReliableSpoolingFileEventReader(
+ new File(dirName), ".COMPLETED",
+ "^$", new File(new File(dirName), ".flumespool"),
+ false, "",
+ "LINE", new Context());
+ } else {
+ reader = new SimpleTextLineEventReader(new InputStreamReader(System.in));
}
- else {
- reader = new BufferedLineReader(new InputStreamReader(System.in));
- }
-
long lastCheck = System.currentTimeMillis();
long sentBytes = 0;
int batchSize = rpcClient.getBatchSize();
- List<String> lines = Lists.newLinkedList();
- while (!(lines = reader.readLines(batchSize)).isEmpty()) {
- List<Event> eventBuffer = Lists.newArrayList();
- for (String line : lines) {
- Event event = EventBuilder.withBody(line, Charsets.UTF_8);
- setHeaders(event);
- eventBuffer.add(event);
+ List<Event> events;
+ while (!(events = reader.readEvents(batchSize)).isEmpty()) {
+ for (Event event : events) {
+ event.setHeaders(headers);
sentBytes += event.getBody().length;
sent++;
@@ -211,7 +204,10 @@ public class AvroCLIClient {
lastCheck = now;
}
}
- rpcClient.appendBatch(eventBuffer);
+ rpcClient.appendBatch(events);
+ if (reader instanceof ReliableEventReader) {
+ ((ReliableEventReader) reader).commit();
+ }
}
logger.debug("Finished");
} finally {
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/BufferedLineReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/BufferedLineReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/BufferedLineReader.java
deleted file mode 100644
index 718e1b2..0000000
--- a/flume-ng-core/src/main/java/org/apache/flume/client/avro/BufferedLineReader.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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.client.avro;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.Reader;
-import java.util.List;
-
-import com.google.common.collect.Lists;
-
-/**
- * A {@link LineReader} implementation which delegates to a
- * {@link BufferedReader}.
- */
-public class BufferedLineReader implements LineReader {
- private BufferedReader reader;
-
- public BufferedLineReader(Reader in) {
- reader = new BufferedReader(in);
- }
-
- @Override
- public String readLine() throws IOException {
- return reader.readLine();
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
- }
-
- @Override
- public List<String> readLines(int n) throws IOException {
- List<String> out = Lists.newLinkedList();
- String line;
- while((line = readLine()) != null && out.size() < n) {
- out.add(line);
- }
- return out;
- }
-
- public void mark(int readAheadLimit) throws IOException {
- reader.mark(readAheadLimit);
- }
-
- public void reset() throws IOException {
- reader.reset();
- }
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/EventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/EventReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/EventReader.java
new file mode 100644
index 0000000..768a381
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/client/avro/EventReader.java
@@ -0,0 +1,58 @@
+/*
+ * 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.client.avro;
+
+import org.apache.flume.Event;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * A line reader produces a stream of lines for the {@link AvroCLIClient} to
+ * ingest into Flume. The stream may be finite or infinite.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface EventReader extends Closeable {
+
+ /**
+ * Get the next line associated with the input stream. If this returns
+ * {@code null}, the input underlying input source is considered finished.
+ * Note that this is allowed to block for indefinite amounts of time waiting
+ * to generate a new line.
+ */
+ public Event readEvent() throws IOException;
+
+ /**
+ * Get up to {@code n} lines associated with the input stream. If this returns
+ * less than n lines, the input underlying input source is considered
+ * finished. Note that this is allowed to block for indefinite amounts of
+ * time waiting to generate a new line.
+ */
+ public List<Event> readEvents(int n) throws IOException;
+
+ /**
+ * Clean-up any state associated with this reader.
+ */
+ 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/client/avro/LineReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/LineReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/LineReader.java
deleted file mode 100644
index 904f22c..0000000
--- a/flume-ng-core/src/main/java/org/apache/flume/client/avro/LineReader.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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.client.avro;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- * A line reader produces a stream of lines for the {@link AvroCLIClient} to
- * ingest into Flume. The stream may be finite or infinite.
- */
-public interface LineReader {
-
- /**
- * Get the next line associated with the input stream. If this returns
- * {@code null}, the input underlying input source is considered finished.
- * Note that this is allowed to block for indefinite amounts of time waiting
- * to generate a new line.
- */
- public String readLine() throws IOException;
-
- /**
- * Get up to {@code n} lines associated with the input stream. If this returns
- * less than n lines, the input underlying input source is considered
- * finished. Note that this is allowed to block for indefinite amounts of
- * time waiting to generate a new line.
- */
- public List<String> readLines(int n) throws IOException;
-
- /**
- * Clean-up any state associated with this reader.
- */
- 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/client/avro/ReliableEventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableEventReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableEventReader.java
new file mode 100644
index 0000000..99c94c6
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableEventReader.java
@@ -0,0 +1,42 @@
+/**
+ * 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.client.avro;
+
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+import java.io.IOException;
+
+/**
+ * A reliable event reader.
+ * Clients must call commit() after each read operation, otherwise the
+ * implementation must reset its internal buffers and return the same events
+ * as it did previously.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface ReliableEventReader extends EventReader {
+
+ /**
+ * Indicate to the implementation that the previously-returned events have
+ * been successfully processed and committed.
+ * @throws IOException
+ */
+ public void commit() throws IOException;
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableSpoolingFileEventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableSpoolingFileEventReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableSpoolingFileEventReader.java
new file mode 100644
index 0000000..f2d587f
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/client/avro/ReliableSpoolingFileEventReader.java
@@ -0,0 +1,415 @@
+/*
+ * 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.client.avro;
+
+import java.io.File;
+import java.io.FileFilter;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.*;
+import java.util.regex.Pattern;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.FlumeException;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.serialization.*;
+import org.apache.flume.tools.PlatformDetect;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Charsets;
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.io.Files;
+
+/**
+ * <p/>A {@link ReliableEventReader} which reads log data from files stored
+ * in a spooling directory and renames each file once all of its data has been
+ * read (through {@link EventDeserializer#readEvent()} calls). The user must
+ * {@link #commit()} each read, to indicate that the lines have been fully
+ * processed.
+ * <p/>Read calls will return no data if there are no files left to read. This
+ * class, in general, is not thread safe.
+ *
+ * <p/>This reader assumes that files with unique file names are left in the
+ * spooling directory and not modified once they are placed there. Any user
+ * behavior which violates these assumptions, when detected, will result in a
+ * FlumeException being thrown.
+ *
+ * <p/>This class makes the following guarantees, if above assumptions are met:
+ * <ul>
+ * <li> Once a log file has been renamed with the {@link #completedSuffix},
+ * all of its records have been read through the
+ * {@link EventDeserializer#readEvent()} function and
+ * {@link #commit()}ed at least once.
+ * <li> All files in the spooling directory will eventually be opened
+ * and delivered to a {@link #readEvents(int)} caller.
+ * </ul>
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class ReliableSpoolingFileEventReader implements ReliableEventReader {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(ReliableSpoolingFileEventReader.class);
+
+ static final String metaFileName = ".flumespool-main.meta";
+
+ private final File directory;
+ private final String completedSuffix;
+ private final String deserializerType;
+ private final Context deserializerContext;
+ private final Pattern ignorePattern;
+ private final File metaFile;
+ private final boolean annotateFileName;
+ private final String fileNameHeader;
+
+ private Optional<FileInfo> currentFile = Optional.absent();
+ /** Always contains the last file from which lines have been read. **/
+ private Optional<FileInfo> lastFileRead = Optional.absent();
+ private boolean committed = true;
+
+ /**
+ * Create a ReliableSpoolingFileEventReader to watch the given directory.
+ */
+ public ReliableSpoolingFileEventReader(File directory, String completedSuffix,
+ String ignorePattern, File trackerDirectory,
+ boolean annotateFileName, String fileNameHeader,
+ String deserializerType, Context deserializerContext) throws IOException {
+
+ // Sanity checks
+ Preconditions.checkNotNull(directory);
+ Preconditions.checkNotNull(completedSuffix);
+ Preconditions.checkNotNull(ignorePattern);
+ Preconditions.checkNotNull(trackerDirectory);
+ Preconditions.checkNotNull(deserializerType);
+ Preconditions.checkNotNull(deserializerContext);
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("Initializing {} with directory={}, metaDir={}, " +
+ "deserializer={}",
+ new Object[] { ReliableSpoolingFileEventReader.class.getSimpleName(),
+ directory, trackerDirectory, deserializerType });
+ }
+
+ // Verify directory exists and is readable/writable
+ Preconditions.checkState(directory.exists(),
+ "Directory does not exist: " + directory.getAbsolutePath());
+ Preconditions.checkState(directory.isDirectory(),
+ "Path is not a directory: " + directory.getAbsolutePath());
+
+ // Do a canary test to make sure we have access to spooling directory
+ try {
+ File f1 = File.createTempFile("flume", "test", directory);
+ Files.write("testing flume file permissions\n", f1, Charsets.UTF_8);
+ Files.readLines(f1, Charsets.UTF_8);
+ if (!f1.delete()) {
+ throw new IOException("Unable to delete canary file " + f1);
+ }
+ } catch (IOException e) {
+ throw new FlumeException("Unable to read and modify files" +
+ " in the spooling directory: " + directory, e);
+ }
+ this.directory = directory;
+ this.completedSuffix = completedSuffix;
+ this.deserializerType = deserializerType;
+ this.deserializerContext = deserializerContext;
+ this.annotateFileName = annotateFileName;
+ this.fileNameHeader = fileNameHeader;
+ this.ignorePattern = Pattern.compile(ignorePattern);
+
+ // ensure that meta directory exists
+ if (!trackerDirectory.exists()) {
+ if (!trackerDirectory.mkdir()) {
+ throw new IOException("Unable to mkdir nonexistent meta directory " +
+ trackerDirectory);
+ }
+ }
+
+ // ensure that the meta directory is a directory
+ if (!trackerDirectory.isDirectory()) {
+ throw new IOException("Specified meta directory is not a directory" +
+ trackerDirectory);
+ }
+
+ this.metaFile = new File(trackerDirectory, metaFileName);
+ }
+
+ /** Return the filename which generated the data from the last successful
+ * {@link #readEvents(int)} call. Returns null if called before any file
+ * contents are read. */
+ public String getLastFileRead() {
+ if (!lastFileRead.isPresent()) {
+ return null;
+ }
+ return lastFileRead.get().getFile().getAbsolutePath();
+ }
+
+ // public interface
+ public Event readEvent() throws IOException {
+ List<Event> events = readEvents(1);
+ if (!events.isEmpty()) {
+ return events.get(0);
+ } else {
+ return null;
+ }
+ }
+
+ public List<Event> readEvents(int numEvents) throws IOException {
+ if (!committed) {
+ if (!currentFile.isPresent()) {
+ throw new IllegalStateException("File should not roll when " +
+ "commit is outstanding.");
+ }
+ logger.info("Last read was never committed - resetting mark position.");
+ currentFile.get().getDeserializer().reset();
+ } else {
+ // Check if new files have arrived since last call
+ if (!currentFile.isPresent()) {
+ currentFile = getNextFile();
+ }
+ // Return empty list if no new files
+ if (!currentFile.isPresent()) {
+ return Collections.emptyList();
+ }
+ }
+
+ EventDeserializer des = currentFile.get().getDeserializer();
+ List<Event> events = des.readEvents(numEvents);
+
+ /* It's possible that the last read took us just up to a file boundary.
+ * If so, try to roll to the next file, if there is one. */
+ if (events.isEmpty()) {
+ retireCurrentFile();
+ currentFile = getNextFile();
+ if (!currentFile.isPresent()) {
+ return Collections.emptyList();
+ }
+ events = currentFile.get().getDeserializer().readEvents(numEvents);
+ }
+
+ if (annotateFileName) {
+ String filename = currentFile.get().getFile().getAbsolutePath();
+ for (Event event : events) {
+ event.getHeaders().put(fileNameHeader, filename);
+ }
+ }
+
+ committed = false;
+ lastFileRead = currentFile;
+ return events;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (currentFile.isPresent()) {
+ currentFile.get().getDeserializer().close();
+ currentFile = Optional.absent();
+ }
+ }
+
+ /** Commit the last lines which were read. */
+ @Override
+ public void commit() throws IOException {
+ if (!committed && currentFile.isPresent()) {
+ currentFile.get().getDeserializer().mark();
+ committed = true;
+ }
+ }
+
+ /**
+ * Closes currentFile and attempt to rename it.
+ *
+ * If these operations fail in a way that may cause duplicate log entries,
+ * an error is logged but no exceptions are thrown. If these operations fail
+ * in a way that indicates potential misuse of the spooling directory, a
+ * FlumeException will be thrown.
+ * @throws FlumeException if files do not conform to spooling assumptions
+ */
+ private void retireCurrentFile() throws IOException {
+ Preconditions.checkState(currentFile.isPresent());
+
+ String currPath = currentFile.get().getFile().getAbsolutePath();
+ String newPath = currPath + completedSuffix;
+ logger.info("Preparing to move file {} to {}", currPath, newPath);
+
+ currentFile.get().getDeserializer().close();
+ File fileToRoll = new File(currPath);
+
+ // Verify that spooling assumptions hold
+ if (fileToRoll.lastModified() != currentFile.get().getLastModified()) {
+ String message = "File has been modified since being read: " + currPath;
+ throw new IllegalStateException(message);
+ }
+ if (fileToRoll.length() != currentFile.get().getLength()) {
+ String message = "File has changed size since being read: " + currPath;
+ throw new IllegalStateException(message);
+ }
+
+ File destination = new File(newPath);
+
+ // Before renaming, check whether destination file name exists
+ if (destination.exists() && PlatformDetect.isWindows()) {
+ /*
+ * If we are here, it means the completed file already exists. In almost
+ * every case this means the user is violating an assumption of Flume
+ * (that log files are placed in the spooling directory with unique
+ * names). However, there is a corner case on Windows systems where the
+ * file was already rolled but the rename was not atomic. If that seems
+ * likely, we let it pass with only a warning.
+ */
+ if (Files.equal(currentFile.get().getFile(), destination)) {
+ logger.warn("Completed file " + newPath +
+ " already exists, but files match, so continuing.");
+ boolean deleted = fileToRoll.delete();
+ if (!deleted) {
+ logger.error("Unable to delete file " + fileToRoll.getAbsolutePath() +
+ ". It will likely be ingested another time.");
+ }
+ } else {
+ String message = "File name has been re-used with different" +
+ " files. Spooling assumptions violated for " + newPath;
+ throw new IllegalStateException(message);
+ }
+
+ // Dest file exists and not on windows
+ } else if (destination.exists()) {
+ String message = "File name has been re-used with different" +
+ " files. Spooling assumptions violated for " + newPath;
+ throw new IllegalStateException(message);
+
+ // Destination file does not already exist. We are good to go!
+ } else {
+ boolean renamed = fileToRoll.renameTo(new File(newPath));
+ if (renamed) {
+ logger.debug("Successfully rolled file {} to {}", fileToRoll, newPath);
+
+ // now we no longer need the meta file
+ deleteMetaFile();
+ } else {
+ /* If we are here then the file cannot be renamed for a reason other
+ * than that the destination file exists (actually, that remains
+ * possible w/ small probability due to TOC-TOU conditions).*/
+ String message = "Unable to move " + currPath + " to " + newPath +
+ ". This will likely cause duplicate events. Please verify that " +
+ "flume has sufficient permissions to perform these operations.";
+ throw new FlumeException(message);
+ }
+ }
+ }
+
+ /**
+ * Find and open the oldest file in the chosen directory. If two or more
+ * files are equally old, the file name with lower lexicographical value is
+ * returned. If the directory is empty, this will return an absent option.
+ */
+ private Optional<FileInfo> getNextFile() {
+ /* Filter to exclude finished or hidden files */
+ FileFilter filter = new FileFilter() {
+ public boolean accept(File candidate) {
+ String fileName = candidate.getName();
+ if ((candidate.isDirectory()) ||
+ (fileName.endsWith(completedSuffix)) ||
+ (fileName.startsWith(".")) ||
+ ignorePattern.matcher(fileName).matches()) {
+ return false;
+ }
+ return true;
+ }
+ };
+ List<File> candidateFiles = Arrays.asList(directory.listFiles(filter));
+ if (candidateFiles.isEmpty()) {
+ return Optional.absent();
+ } else {
+ Collections.sort(candidateFiles, new Comparator<File>() {
+ public int compare(File a, File b) {
+ int timeComparison = new Long(a.lastModified()).compareTo(
+ new Long(b.lastModified()));
+ if (timeComparison != 0) {
+ return timeComparison;
+ }
+ else {
+ return a.getName().compareTo(b.getName());
+ }
+ }
+ });
+ File nextFile = candidateFiles.get(0);
+ try {
+ // roll the meta file, if needed
+ String nextPath = nextFile.getPath();
+ PositionTracker tracker =
+ DurablePositionTracker.getInstance(metaFile, nextPath);
+ if (!tracker.getTarget().equals(nextPath)) {
+ tracker.close();
+ deleteMetaFile();
+ }
+ tracker = DurablePositionTracker.getInstance(metaFile, nextPath);
+
+ // sanity check
+ Preconditions.checkState(tracker.getTarget().equals(nextPath),
+ "Tracker target %s does not equal expected filename %s",
+ tracker.getTarget(), nextPath);
+
+ ResettableInputStream in =
+ new ResettableFileInputStream(nextFile, tracker);
+ EventDeserializer deserializer = EventDeserializerFactory.getInstance
+ (deserializerType, deserializerContext, in);
+
+ return Optional.of(new FileInfo(nextFile, deserializer));
+ } catch (FileNotFoundException e) {
+ // File could have been deleted in the interim
+ logger.warn("Could not find file: " + nextFile, e);
+ return Optional.absent();
+ } catch (IOException e) {
+ logger.error("Exception opening file: " + nextFile, e);
+ return Optional.absent();
+ }
+ }
+ }
+
+ private void deleteMetaFile() throws IOException {
+ if (metaFile.exists() && !metaFile.delete()) {
+ throw new IOException("Unable to delete old meta file " + metaFile);
+ }
+ }
+
+ /** An immutable class with information about a file being processed. */
+ private static class FileInfo {
+ private final File file;
+ private final long length;
+ private final long lastModified;
+ private final EventDeserializer deserializer;
+
+ public FileInfo(File file, EventDeserializer deserializer) {
+ this.file = file;
+ this.length = file.length();
+ this.lastModified = file.lastModified();
+ this.deserializer = deserializer;
+ }
+
+ public long getLength() { return length; }
+ public long getLastModified() { return lastModified; }
+ public EventDeserializer getDeserializer() { return deserializer; }
+ public File getFile() { return file; }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/SimpleTextLineEventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/SimpleTextLineEventReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/SimpleTextLineEventReader.java
new file mode 100644
index 0000000..2fcb152
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/client/avro/SimpleTextLineEventReader.java
@@ -0,0 +1,76 @@
+/*
+ * 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.client.avro;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.Reader;
+import java.util.List;
+
+import com.google.common.base.Charsets;
+import com.google.common.collect.Lists;
+import org.apache.flume.Event;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.EventBuilder;
+
+/**
+ * A {@link EventReader} implementation which delegates to a
+ * {@link BufferedReader}.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class SimpleTextLineEventReader implements EventReader {
+
+ private final BufferedReader reader;
+
+ public SimpleTextLineEventReader(Reader in) {
+ reader = new BufferedReader(in);
+ }
+
+ @Override
+ public Event readEvent() throws IOException {
+ String line = reader.readLine();
+ if (line != null) {
+ return EventBuilder.withBody(line, Charsets.UTF_8);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public List<Event> readEvents(int n) throws IOException {
+ List<Event> events = Lists.newLinkedList();
+ while (events.size() < n) {
+ Event event = readEvent();
+ if (event != null) {
+ events.add(event);
+ } else {
+ break;
+ }
+ }
+ return events;
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/client/avro/SpoolingFileLineReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/client/avro/SpoolingFileLineReader.java b/flume-ng-core/src/main/java/org/apache/flume/client/avro/SpoolingFileLineReader.java
deleted file mode 100644
index 8362299..0000000
--- a/flume-ng-core/src/main/java/org/apache/flume/client/avro/SpoolingFileLineReader.java
+++ /dev/null
@@ -1,361 +0,0 @@
-/*
- * 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.client.avro;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileFilter;
-import java.io.FileNotFoundException;
-import java.io.FileReader;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.List;
-
-import org.apache.flume.FlumeException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Charsets;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-import com.google.common.io.Files;
-
-/**
- * A {@link LineReader} which reads log data from files stored in a
- * spooling directory and renames each file once all of its data has been
- * read (through {@link #readLine()} calls). The user must {@link #commit()}
- * each read, to indicate that the lines have been fully processed.
- * Read calls will return no data if there are no files left to read. This
- * class, in general, is not thread safe.
- *
- * This reader assumes that files with unique file names are left in the
- * spooling directory and not modified once they are placed there. Any user
- * behavior which violates these assumptions, when detected, will result in a
- * FlumeException being thrown.
- *
- * This class makes the following guarantees, if above assumptions are met:
- * 1) Once a log file has been renamed with the {@link #completedSuffix},
- * all of its records have been read through the {@link #readLine()}
- * function and {@link #commit()}'d exactly once.
- * 2) All files in the spooling directory will eventually be opened
- * and delivered to a {@link #readLine()} caller.
- *
- * NOTE: This class is for internal Flume components, this is not an extension
- * point for developers customizing Flume.
- */
-public class SpoolingFileLineReader implements LineReader {
- private static final Logger logger = LoggerFactory
- .getLogger(SpoolingFileLineReader.class);
- /** When a line is too long, how many characters do we print to err log. **/
- private static final int OVERFLOW_LINE_PRINT_CHARS = 30;
-
- private File directory;
- public String completedSuffix;
- private Optional<FileInfo> currentFile = Optional.absent();
- /** Always contains the last file from which lines have been read. **/
- private Optional<FileInfo> lastFileRead = Optional.absent();
- private boolean committed = true;
- private int bufferMaxLines;
- private int bufferMaxLineLength;
- /** A flag to signal an un-recoverable error has occured. */
- private boolean disabled = false;
-
- /**
- * Create a SpoolingFileLineReader to watch the given directory.
- *
- * Lines are buffered between when they are read and when they are committed.
- * The buffer has a fixed size. Its size is determined by (maxLinestoBuffer *
- * bufferSizePerLine).
- *
- * @param directory The directory to watch
- * @param completedSuffix The suffix to append to completed files
- * @param bufferMaxLines The maximum number of lines to keep in a pre-commit
- * buffer
- * @param bufferMaxLineLength The maximum line length for lines in the pre-commit
- * buffer, in characters
- */
- public SpoolingFileLineReader(File directory, String completedSuffix,
- int bufferMaxLines, int bufferMaxLineLength) {
- // Verify directory exists and is readable/writable
- Preconditions.checkNotNull(directory);
- Preconditions.checkState(directory.exists(),
- "Directory does not exist: " + directory.getAbsolutePath());
- Preconditions.checkState(directory.isDirectory(),
- "Path is not a directory: " + directory.getAbsolutePath());
- Preconditions.checkState(bufferMaxLines > 0);
- Preconditions.checkState(bufferMaxLineLength > 0);
-
- // Do a canary test to make sure we have access to spooling directory
- try {
- File f1 = File.createTempFile("flume", "test", directory);
- Files.write("testing flume file permissions\n", f1, Charsets.UTF_8);
- Files.readLines(f1, Charsets.UTF_8);
- f1.delete();
- } catch (IOException e) {
- throw new FlumeException("Unable to read and modify files" +
- " in the spooling directory: " + directory, e);
- }
- this.directory = directory;
- this.completedSuffix = completedSuffix;
- this.bufferMaxLines = bufferMaxLines;
- this.bufferMaxLineLength = bufferMaxLineLength;
- }
-
- /** Return the filename which generated the data from the last successful
- * {@link #readLines()} call. Returns null if called before any file
- * contents are read. */
- public String getLastFileRead() {
- if (!lastFileRead.isPresent()) {
- return null;
- }
- return lastFileRead.get().getFile().getAbsolutePath();
- }
-
- @Override
- public String readLine() throws IOException {
- if (disabled) {
- throw new IllegalStateException("Reader has been disabled.");
- }
- List<String> read = readLines(1);
- if (read.size() == 0) {
- return null;
- }
- return read.get(0);
- }
-
- /** Commit the last lines which were read. */
- public void commit() throws IOException {
- if (disabled) {
- throw new IllegalStateException("Reader has been disabled.");
- }
- currentFile.get().reader.mark(bufferMaxLines * bufferMaxLineLength);
- committed = true;
- }
-
- @Override
- /** Reads up to n lines from the current file. Returns an empty list if no
- * files are left to read in the directory. */
- public List<String> readLines(int n) throws IOException {
- if (disabled) {
- throw new IllegalStateException("Reader has been disabled.");
- }
- if (!committed) {
- if (!currentFile.isPresent()) {
- throw new IllegalStateException("File should not roll when " +
- " commit is outstanding.");
- }
- logger.info("Last read was never comitted - resetting mark position.");
- currentFile.get().getReader().reset();
- } else {
- // Check if new files have arrived since last call
- if (!currentFile.isPresent()) {
- currentFile = getNextFile();
- }
- // Return empty list if no new files
- if (!currentFile.isPresent()) {
- return Collections.emptyList();
- }
- }
-
- String outLine = currentFile.get().getReader().readLine();
-
- /* It's possible that the last read took us just up to a file boundary.
- * If so, try to roll to the next file, if there is one. */
- if (outLine == null) {
- retireCurrentFile();
- currentFile = getNextFile();
- if (!currentFile.isPresent()) {
- return Collections.emptyList();
- }
- outLine = currentFile.get().getReader().readLine();
- }
-
- List<String> out = Lists.newArrayList();
- while (outLine != null) {
- if (outLine.length() > bufferMaxLineLength) {
- logger.error("Found line longer than " + bufferMaxLineLength +
- " characters, cannot make progress.");
- int lastCharToPrint = Math.min(OVERFLOW_LINE_PRINT_CHARS,
- outLine.length());
- logger.error("Invalid line starts with: " +
- outLine.substring(0, lastCharToPrint));
- disabled = true;
- throw new FlumeException("Encoutered line that was too long.");
- }
- out.add(outLine);
- if (out.size() == n) { break; }
- outLine = currentFile.get().getReader().readLine();
- }
-
- committed = false;
- lastFileRead = currentFile;
- return out;
- }
-
- /**
- * Closes currentFile and attempt to rename it.
- *
- * If these operations fail in a way that may cause duplicate log entries,
- * an error is logged but no exceptions are thrown. If these operations fail
- * in a way that indicates potential misuse of the spooling directory, a
- * FlumeException will be thrown.
- * @throws FlumeException if files do not conform to spooling assumptions
- */
- private void retireCurrentFile() throws IOException {
- Preconditions.checkState(currentFile.isPresent());
-
- String currPath = currentFile.get().getFile().getAbsolutePath();
- String newPath = currPath + completedSuffix;
- logger.info("Preparing to move file " + currPath + " to " + newPath);
-
- currentFile.get().reader.close();
- File newFile = new File(currPath);
-
- // Verify that spooling assumptions hold
- if (newFile.lastModified() != currentFile.get().getLastModified()) {
- String message = "File has been modified since being read: " + currPath;
- throw new IllegalStateException(message);
- }
- if (newFile.length() != currentFile.get().getLength()) {
- String message = "File has changed size since being read: " + currPath;
- throw new IllegalStateException(message);
- }
-
- // Before renaming, check whether destination file name exists
- File existing = new File(newPath);
- if (existing.exists() &&
- System.getProperty("os.name").toLowerCase().indexOf("win") >= 0) {
- /*
- * If we are here, it means the completed file already exists. In almost
- * every case this means the user is violating an assumption of Flume
- * (that log files are placed in the spooling directory with unique
- * names). However, there is a corner case on Windows systems where the
- * file was already rolled but the rename was not atomic. If that seems
- * likely, we let it pass with only a warning.
- */
- if (Files.equal(currentFile.get().getFile(), existing)) {
- logger.warn("Completed file " + newPath +
- " already exists, but files match, so continuing.");
- boolean deleted = newFile.delete();
- if (!deleted) {
- logger.error("Unable to delete file " + newFile.getAbsolutePath() +
- ". It will likely be ingested another time.");
- }
- } else {
- String message = "File name has been re-used with different" +
- " files. Spooling assumption violated for " + newPath;
- throw new IllegalStateException(message);
- }
- } else if (existing.exists()) { // Dest file exists and not on windows
- String message = "File name has been re-used with different" +
- " files. Spooling assumption violated for " + newPath;
- throw new IllegalStateException(message);
- } else { // Dest file does not already exist
- boolean renamed = newFile.renameTo(new File(newPath));
- if (!renamed) {
- /* If we are here then the file cannot be renamed for a reason other
- * than that the destination file exists (actually, that remains
- * possible w/ small probability due to TOC-TOU conditions).*/
- String message = "Unable to move " + currPath + " to " + newPath +
- ". This will likely cause duplicate events. Please verify that " +
- "flume has sufficient permissions to perform these operations.";
- throw new FlumeException(message);
- }
- }
- }
-
- /**
- * Find and open the oldest file in the chosen directory. If two or more
- * files are equally old, the file name with lower lexicographical value is
- * returned. If the directory is empty, this will return an absent option.
- */
- private Optional<FileInfo> getNextFile() {
- /* Filter to exclude finished or hidden files */
- FileFilter filter = new FileFilter() {
- public boolean accept(File pathName) {
- if ((pathName.getName().endsWith(completedSuffix)) ||
- (pathName.getName().startsWith("."))) {
- return false;
- }
- return true;
- }
- };
- List<File> candidateFiles = Arrays.asList(directory.listFiles(filter));
- if (candidateFiles.isEmpty()) {
- return Optional.absent();
- } else {
- Collections.sort(candidateFiles, new Comparator<File>() {
- public int compare(File a, File b) {
- int timeComparison = new Long(a.lastModified()).compareTo(
- new Long(b.lastModified()));
- if (timeComparison != 0) {
- return timeComparison;
- }
- else {
- return a.getName().compareTo(b.getName());
- }
- }
- });
- File nextFile = candidateFiles.get(0);
- try {
- int bufferSize = bufferMaxLines * bufferMaxLineLength;
- BufferedReader reader = new BufferedReader(new FileReader(nextFile),
- bufferSize);
- reader.mark(bufferSize);
- return Optional.of(new FileInfo(nextFile, reader));
- } catch (FileNotFoundException e) {
- // File could have been deleted in the interim
- logger.warn("Could not find file: " + nextFile, e);
- return Optional.absent();
- } catch (IOException e) {
- logger.error("Exception opening file: " + nextFile, e);
- return Optional.absent();
- }
- }
- }
-
- /** An immutable class with information about a file being processed. */
- private class FileInfo {
- private File file;
- private long length;
- private long lastModified;
- private BufferedReader reader;
-
- public FileInfo(File file, BufferedReader reader) {
- this.file = file;
- this.length = file.length();
- this.lastModified = file.lastModified();
- this.reader = reader;
- }
-
- public long getLength() { return this.length; }
- public long getLastModified() { return this.lastModified; }
- public BufferedReader getReader() { return this.reader; }
- public File getFile() { return this.file; }
- }
-
- @Override
- public void close() throws IOException {
- // No-op
- }
-}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/DurablePositionTracker.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/DurablePositionTracker.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/DurablePositionTracker.java
new file mode 100644
index 0000000..a4511a7
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/DurablePositionTracker.java
@@ -0,0 +1,202 @@
+/**
+ * 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.File;
+import java.io.IOException;
+
+import com.google.common.base.Preconditions;
+import org.apache.avro.file.DataFileReader;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.io.DatumReader;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.tools.PlatformDetect;
+
+/**
+ * <p/>Class that stores object state in an avro container file.
+ * The file is only ever appended to.
+ * At construction time, the object reads data from the end of the file and
+ * caches that data for use by a client application. After construction, reads
+ * never go to disk.
+ * Writes always flush to disk.
+ *
+ * <p/>Note: This class is not thread-safe.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class DurablePositionTracker implements PositionTracker {
+
+ private final File trackerFile;
+ private final DataFileWriter<TransferStateFileMeta> writer;
+ private final DataFileReader<TransferStateFileMeta> reader;
+ private final TransferStateFileMeta metaCache;
+
+ private String target;
+
+ private boolean isOpen;
+
+ /**
+ * If the file exists at startup, then read it, roll it, and open a new one.
+ * We go through this to avoid issues with partial reads at the end of the
+ * file from a previous crash. If we append to a bad record,
+ * our writes may never be visible.
+ * @param trackerFile
+ * @param target
+ * @return
+ * @throws IOException
+ */
+ public static DurablePositionTracker getInstance(File trackerFile,
+ String target) throws IOException {
+
+ if (!trackerFile.exists()) {
+ return new DurablePositionTracker(trackerFile, target);
+ }
+
+ // exists
+ DurablePositionTracker oldTracker =
+ new DurablePositionTracker(trackerFile, target);
+ String existingTarget = oldTracker.getTarget();
+ long targetPosition = oldTracker.getPosition();
+ oldTracker.close();
+
+ File tmpMeta = File.createTempFile(trackerFile.getName(), ".tmp",
+ trackerFile.getParentFile());
+ tmpMeta.delete();
+ DurablePositionTracker tmpTracker =
+ new DurablePositionTracker(tmpMeta, existingTarget);
+ tmpTracker.storePosition(targetPosition);
+ tmpTracker.close();
+
+ // On windows, things get messy with renames...
+ // FIXME: This is not atomic. Consider implementing a recovery procedure
+ // so that if it does not exist at startup, check for a rolled version
+ // before creating a new file from scratch.
+ if (PlatformDetect.isWindows()) {
+ if (!trackerFile.delete()) {
+ throw new IOException("Unable to delete existing meta file " +
+ trackerFile);
+ }
+ }
+
+ // rename tmp file to meta
+ if (!tmpMeta.renameTo(trackerFile)) {
+ throw new IOException("Unable to rename " + tmpMeta + " to " +
+ trackerFile);
+ }
+
+ // return a new known-good version that is open for append
+ DurablePositionTracker newTracker =
+ new DurablePositionTracker(trackerFile, existingTarget);
+ return newTracker;
+ }
+
+
+ /**
+ * If the file exists, read it and open it for append.
+ * @param trackerFile
+ * @param target
+ * @throws IOException
+ */
+ DurablePositionTracker(File trackerFile, String target)
+ throws IOException {
+
+ Preconditions.checkNotNull(trackerFile, "trackerFile must not be null");
+ Preconditions.checkNotNull(target, "target must not be null");
+
+ this.trackerFile = trackerFile;
+ this.target = target;
+
+ DatumWriter<TransferStateFileMeta> dout =
+ new SpecificDatumWriter<TransferStateFileMeta>(
+ TransferStateFileMeta.SCHEMA$);
+
+ DatumReader<TransferStateFileMeta> din =
+ new SpecificDatumReader<TransferStateFileMeta>(
+ TransferStateFileMeta.SCHEMA$);
+
+ writer = new DataFileWriter<TransferStateFileMeta>(dout);
+
+ if (trackerFile.exists()) {
+ // open it for append
+ writer.appendTo(trackerFile);
+
+ reader = new DataFileReader<TransferStateFileMeta>(trackerFile, din);
+ this.target = reader.getMetaString("file");
+ } else {
+ // create the file
+ this.target = target;
+ writer.setMeta("file", target);
+ writer.create(TransferStateFileMeta.SCHEMA$, trackerFile);
+ reader = new DataFileReader<TransferStateFileMeta>(trackerFile, din);
+ }
+
+ target = getTarget();
+
+ // initialize @ line = 0;
+ metaCache = TransferStateFileMeta.newBuilder().setOffset(0L).build();
+
+ initReader();
+
+ isOpen = true;
+ }
+
+ /**
+ * Read the last record in the file.
+ */
+ private void initReader() throws IOException {
+ long syncPos = trackerFile.length() - 256L;
+ if (syncPos < 0) syncPos = 0L;
+ reader.sync(syncPos);
+ while (reader.hasNext()) {
+ reader.next(metaCache);
+ }
+ }
+
+ @Override
+ public synchronized void storePosition(long position) throws IOException {
+ metaCache.setOffset(position);
+ writer.append(metaCache);
+ writer.sync();
+ writer.flush();
+ }
+
+ @Override
+ public synchronized long getPosition() {
+ return metaCache.getOffset();
+ }
+
+ @Override
+ public String getTarget() {
+ return target;
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (isOpen) {
+ writer.close();
+ reader.close();
+ isOpen = false;
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializer.java
new file mode 100644
index 0000000..bbb3375
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializer.java
@@ -0,0 +1,94 @@
+/**
+ * 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.Closeable;
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+/**
+ * Establishes a contract for reading events stored in arbitrary formats from
+ * reliable, resettable streams.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface EventDeserializer extends Resettable, Closeable {
+
+ /**
+ * Read a single event from the underlying stream.
+ * @return Deserialized event or {@code null} if no events could be read.
+ * @throws IOException
+ * @see #mark()
+ * @see #reset()
+ */
+ public Event readEvent() throws IOException;
+
+ /**
+ * Read a batch of events from the underlying stream.
+ * @param numEvents Maximum number of events to return.
+ * @return List of read events, or empty list if no events could be read.
+ * @throws IOException
+ * @see #mark()
+ * @see #reset()
+ */
+ public List<Event> readEvents(int numEvents) throws IOException;
+
+ /**
+ * Marks the underlying input stream, indicating that the events previously
+ * returned by this EventDeserializer have been successfully committed.
+ * @throws IOException
+ * @see #reset()
+ */
+ @Override
+ public void mark() throws IOException;
+
+ /**
+ * Resets the underlying input stream to the last known mark (or beginning
+ * of the stream if {@link #mark()} was never previously called. This should
+ * be done in the case of inability to commit previously-deserialized events.
+ * @throws IOException
+ * @see #mark()
+ */
+ @Override
+ public void reset() throws IOException;
+
+ /**
+ * Calls {@link #reset()} on the stream and then closes it.
+ * In the case of successful completion of event consumption,
+ * {@link #mark()} MUST be called before {@code close()}.
+ * @throws IOException
+ * @see #mark()
+ * @see #reset()
+ */
+ @Override
+ public void close() throws IOException;
+
+ /**
+ * Knows how to construct this deserializer.<br/>
+ * <b>Note: Implementations MUST provide a public a no-arg constructor.</b>
+ */
+ public interface Builder {
+ public EventDeserializer build(Context context, ResettableInputStream in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerFactory.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerFactory.java
new file mode 100644
index 0000000..80b9cc3
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerFactory.java
@@ -0,0 +1,88 @@
+/*
+ * 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.Preconditions;
+import org.apache.flume.Context;
+import org.apache.flume.FlumeException;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@InterfaceAudience.Private
+@InterfaceStability.Stable
+public class EventDeserializerFactory {
+
+ private static final Logger logger =
+ LoggerFactory.getLogger(EventDeserializerFactory.class);
+
+ public static EventDeserializer getInstance(
+ String deserializerType, Context context, ResettableInputStream in) {
+
+ Preconditions.checkNotNull(deserializerType,
+ "serializer type must not be null");
+
+ // try to find builder class in enum of known output serializers
+ EventDeserializerType type;
+ try {
+ type = EventDeserializerType.valueOf(deserializerType.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ logger.debug("Not in enum, loading builder class: {}", deserializerType);
+ type = EventDeserializerType.OTHER;
+ }
+ Class<? extends EventDeserializer.Builder> builderClass =
+ type.getBuilderClass();
+
+ // handle the case where they have specified their own builder in the config
+ if (builderClass == null) {
+ try {
+ Class c = Class.forName(deserializerType);
+ if (c != null && EventDeserializer.Builder.class.isAssignableFrom(c)) {
+ builderClass = (Class<? extends EventDeserializer.Builder>) c;
+ } else {
+ String errMessage = "Unable to instantiate Builder from " +
+ deserializerType + ": does not appear to implement " +
+ EventDeserializer.Builder.class.getName();
+ throw new FlumeException(errMessage);
+ }
+ } catch (ClassNotFoundException ex) {
+ logger.error("Class not found: " + deserializerType, ex);
+ throw new FlumeException(ex);
+ }
+ }
+
+ // build the builder
+ EventDeserializer.Builder builder;
+ try {
+ builder = builderClass.newInstance();
+ } catch (InstantiationException ex) {
+ String errMessage = "Cannot instantiate builder: " + deserializerType;
+ logger.error(errMessage, ex);
+ throw new FlumeException(errMessage, ex);
+ } catch (IllegalAccessException ex) {
+ String errMessage = "Cannot instantiate builder: " + deserializerType;
+ logger.error(errMessage, ex);
+ throw new FlumeException(errMessage, ex);
+ }
+
+ return builder.build(context, in);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerType.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerType.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerType.java
new file mode 100644
index 0000000..02abc80
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventDeserializerType.java
@@ -0,0 +1,40 @@
+/*
+ * 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;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public enum EventDeserializerType {
+ LINE(LineDeserializer.Builder.class),
+ OTHER(null);
+
+ private final Class<? extends EventDeserializer.Builder> builderClass;
+
+ EventDeserializerType(Class<? extends EventDeserializer.Builder> builderClass) {
+ this.builderClass = builderClass;
+ }
+
+ public Class<? extends EventDeserializer.Builder> getBuilderClass() {
+ return builderClass;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerDe.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerDe.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerDe.java
new file mode 100644
index 0000000..fb1663a
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerDe.java
@@ -0,0 +1,49 @@
+/**
+ * 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.OutputStream;
+import org.apache.flume.Context;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+/**
+ * <p/>A class that is able to both serialize and deserialize events.
+ * <p/>Implementing this interface does not simply mean that a class has both
+ * serialization and deserialization capability. By implementing this
+ * interface, implementations guarantee that they can convert a serialized
+ * event back to a deserialized event, and back to a serialized event again
+ * without any data loss.
+ * <p/>That guarantee allows Flume to "replay" partial file writes and restore
+ * output files that may have been damaged in a system crash. At the time of
+ * this writing, support for this functionality is still lacking.
+ */
+@InterfaceAudience.LimitedPrivate("future public api")
+@InterfaceStability.Unstable
+public interface EventSerDe extends EventSerializer, EventDeserializer {
+
+ /**
+ * Knows how to construct this serde.<br/>
+ * <b>Note: Implementations MUST provide a public a no-arg constructor.</b>
+ */
+ public interface Builder {
+ public EventSerDe build(Context context, ResettableInputStream in, OutputStream out);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
index a418935..e230e34 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializer.java
@@ -127,7 +127,7 @@ public interface EventSerializer {
public boolean supportsReopen();
/**
- * Knows how to construct this output formatter.<br/>
+ * Knows how to construct this event serializer.<br/>
* <b>Note: Implementations MUST provide a public a no-arg constructor.</b>
*/
public interface Builder {
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerFactory.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerFactory.java
index 75853a9..dbb3935 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerFactory.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerFactory.java
@@ -22,9 +22,13 @@ import com.google.common.base.Preconditions;
import java.io.OutputStream;
import org.apache.flume.Context;
import org.apache.flume.FlumeException;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@InterfaceAudience.Private
+@InterfaceStability.Stable
public class EventSerializerFactory {
private static final Logger logger =
@@ -55,8 +59,8 @@ public class EventSerializerFactory {
builderClass = (Class<? extends EventSerializer.Builder>) c;
} else {
String errMessage = "Unable to instantiate Builder from " +
- serializerType;
- logger.error(errMessage);
+ serializerType + ": does not appear to implement " +
+ EventSerializer.Builder.class.getName();
throw new FlumeException(errMessage);
}
} catch (ClassNotFoundException ex) {
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerType.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerType.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerType.java
index afe8ed8..450b96f 100644
--- a/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerType.java
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/EventSerializerType.java
@@ -18,6 +18,11 @@
*/
package org.apache.flume.serialization;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
public enum EventSerializerType {
TEXT(BodyTextEventSerializer.Builder.class),
AVRO_EVENT(FlumeEventAvroEventSerializer.Builder.class),
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/LineDeserializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/LineDeserializer.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/LineDeserializer.java
new file mode 100644
index 0000000..d8e5ae2
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/LineDeserializer.java
@@ -0,0 +1,167 @@
+/**
+ * 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.collect.Lists;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.annotations.InterfaceAudience;
+import org.apache.flume.annotations.InterfaceStability;
+import org.apache.flume.event.EventBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * A deserializer that parses text lines from a file.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class LineDeserializer implements EventDeserializer {
+
+ private static final Logger logger = LoggerFactory.getLogger
+ (LineDeserializer.class);
+
+ private final ResettableInputStream in;
+ private final Charset outputCharset;
+ private final int maxLineLength;
+ private volatile boolean isOpen;
+
+ public static final String OUT_CHARSET_KEY = "outputCharset";
+ public static final String CHARSET_DFLT = "UTF-8";
+
+ public static final String MAXLINE_KEY = "maxLineLength";
+ public static final int MAXLINE_DFLT = 2048;
+
+ LineDeserializer(Context context, ResettableInputStream in) {
+ this.in = in;
+ this.outputCharset = Charset.forName(
+ context.getString(OUT_CHARSET_KEY, CHARSET_DFLT));
+ this.maxLineLength = context.getInteger(MAXLINE_KEY, MAXLINE_DFLT);
+ this.isOpen = true;
+ }
+
+ /**
+ * Reads a line from a file and returns an event
+ * @return Event containing parsed line
+ * @throws IOException
+ */
+ @Override
+ public Event readEvent() throws IOException {
+ ensureOpen();
+ String line = readLine();
+ if (line == null) {
+ return null;
+ } else {
+ return EventBuilder.withBody(line, outputCharset);
+ }
+ }
+
+ /**
+ * Batch line read
+ * @param numEvents Maximum number of events to return.
+ * @return List of events containing read lines
+ * @throws IOException
+ */
+ @Override
+ public List<Event> readEvents(int numEvents) throws IOException {
+ ensureOpen();
+ List<Event> events = Lists.newLinkedList();
+ for (int i = 0; i < numEvents; i++) {
+ Event event = readEvent();
+ if (event != null) {
+ events.add(event);
+ } else {
+ break;
+ }
+ }
+ return events;
+ }
+
+ @Override
+ public void mark() throws IOException {
+ ensureOpen();
+ in.mark();
+ }
+
+ @Override
+ public void reset() throws IOException {
+ ensureOpen();
+ in.reset();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (isOpen) {
+ reset();
+ in.close();
+ isOpen = false;
+ }
+ }
+
+ private void ensureOpen() {
+ if (!isOpen) {
+ throw new IllegalStateException("Serializer has been closed");
+ }
+ }
+
+ // TODO: consider not returning a final character that is a high surrogate
+ // when truncating
+ private String readLine() throws IOException {
+ StringBuilder sb = new StringBuilder();
+ int c;
+ int readChars = 0;
+ while ((c = in.readChar()) != -1) {
+ readChars++;
+
+ // FIXME: support \r\n
+ if (c == '\n') {
+ break;
+ }
+
+ sb.append((char)c);
+
+ if (readChars >= maxLineLength) {
+ logger.warn("Line length exceeds max ({}), truncating line!",
+ maxLineLength);
+ break;
+ }
+ }
+
+ if (readChars > 0) {
+ return sb.toString();
+ } else {
+ return null;
+ }
+ }
+
+ public static class Builder implements EventDeserializer.Builder {
+
+ @Override
+ public EventDeserializer build(Context context, ResettableInputStream in) {
+ return new LineDeserializer(context, in);
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/PositionTracker.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/PositionTracker.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/PositionTracker.java
new file mode 100644
index 0000000..ff46133
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/PositionTracker.java
@@ -0,0 +1,36 @@
+/**
+ * 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;
+
+/**
+ * Defines an interface for tracking the offset position in a target file.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface PositionTracker extends Closeable {
+ void storePosition(long position) throws IOException;
+ long getPosition();
+ String getTarget();
+ void close() throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/flume/blob/fad7251d/flume-ng-core/src/main/java/org/apache/flume/serialization/Resettable.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/main/java/org/apache/flume/serialization/Resettable.java b/flume-ng-core/src/main/java/org/apache/flume/serialization/Resettable.java
new file mode 100644
index 0000000..d56e311
--- /dev/null
+++ b/flume-ng-core/src/main/java/org/apache/flume/serialization/Resettable.java
@@ -0,0 +1,47 @@
+/**
+ * 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.IOException;
+
+/**
+ * Defines an API for objects that can be mark()ed and reset() on arbitrary
+ * boundaries. Any implementation that has a limited buffer for the mark(),
+ * like {@link java.io.InputStream}, must not implement Resettable.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface Resettable {
+
+ /**
+ * Indicate that the current position should be returned to in the case of
+ * {@link #reset()} being called.
+ * @throws IOException
+ */
+ void mark() throws IOException;
+
+ /**
+ * Return to the last marked position, or the beginning of the stream if
+ * {@link #mark()} has never been called.
+ * @throws IOException
+ */
+ void reset() throws IOException;
+}