Enhance taildir source's efficiencies

flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/ReliableTaildirEventReader.java

@@ -49,7 +49,7 @@
 public class ReliableTaildirEventReader implements ReliableEventReader {
   private static final Logger logger = LoggerFactory.getLogger(ReliableTaildirEventReader.class);
 
-  private final List<TaildirMatcher> taildirCache;
+  private final List<TailMatcher> taildirCache;
   private final Table<String, String, String> headerTable;
 
   private TailFile currentFile = null;
@@ -65,22 +65,34 @@ public class ReliableTaildirEventReader implements ReliableEventReader {
    * Create a ReliableTaildirEventReader to watch the given directory.
    */
   private ReliableTaildirEventReader(Map<String, String> filePaths,
+      Map<String, String> filePathsIncludeChild,
       Table<String, String, String> headerTable, String positionFilePath,
       boolean skipToEnd, boolean addByteOffset, boolean cachePatternMatching,
       boolean annotateFileName, String fileNameHeader) throws IOException {
     // Sanity checks
-    Preconditions.checkNotNull(filePaths);
+    if (filePaths == null && filePathsIncludeChild == null) {
+      throw new NullPointerException();
+    }
     Preconditions.checkNotNull(positionFilePath);
 
     if (logger.isDebugEnabled()) {
       logger.debug("Initializing {} with directory={}",
-          new Object[] { ReliableTaildirEventReader.class.getSimpleName(), filePaths });
+              new Object[]{ReliableTaildirEventReader.class.getSimpleName(), filePaths});
     }
 
-    List<TaildirMatcher> taildirCache = Lists.newArrayList();
-    for (Entry<String, String> e : filePaths.entrySet()) {
-      taildirCache.add(new TaildirMatcher(e.getKey(), e.getValue(), cachePatternMatching));
+    List<TailMatcher> taildirCache = Lists.newArrayList();
+    if (filePaths != null) {
+      for (Entry<String, String> e : filePaths.entrySet()) {
+        taildirCache.add(new TaildirMatcher(e.getKey(), e.getValue(), cachePatternMatching));
+      }
+    }
+    if (filePathsIncludeChild != null) {
+      for (Map.Entry<String, String> e : filePathsIncludeChild.entrySet()) {
+        taildirCache.add(
+                new TaildirIncludeChildMatcher(e.getKey(), e.getValue(), cachePatternMatching));
+      }
     }
+
     logger.info("taildirCache: " + taildirCache.toString());
     logger.info("headerTable: " + headerTable.toString());
 
@@ -187,7 +199,7 @@ public List<Event> readEvents(int numEvents, boolean backoffWithoutNL)
       throws IOException {
     if (!committed) {
       if (currentFile == null) {
-        throw new IllegalStateException("current file does not exist. " + currentFile.getPath());
+        throw new IllegalStateException("current file does not exist. ");
       }
       logger.info("Last read was never committed - resetting position");
       long lastPos = currentFile.getPos();
@@ -239,14 +251,15 @@ public List<Long> updateTailFiles(boolean skipToEnd) throws IOException {
     updateTime = System.currentTimeMillis();
     List<Long> updatedInodes = Lists.newArrayList();
 
-    for (TaildirMatcher taildir : taildirCache) {
+    for (TailMatcher taildir : taildirCache) {
       Map<String, String> headers = headerTable.row(taildir.getFileGroup());
 
       for (File f : taildir.getMatchingFiles()) {
         long inode;
         try {
           inode = getInode(f);
         } catch (NoSuchFileException e) {
+          taildir.deleteFileCache(f);
           logger.info("File has been deleted in the meantime: " + e.getMessage());
           continue;
         }
@@ -299,6 +312,7 @@ private TailFile openFile(File file, Map<String, String> headers, long inode, lo
    */
   public static class Builder {
     private Map<String, String> filePaths;
+    private Map<String, String> filePathsIncludeChild;
     private Table<String, String, String> headerTable;
     private String positionFilePath;
     private boolean skipToEnd;
@@ -314,6 +328,11 @@ public Builder filePaths(Map<String, String> filePaths) {
       return this;
     }
 
+    public Builder filePathsIncludeChild(Map<String, String> filePathsIncludeChild) {
+      this.filePathsIncludeChild = filePathsIncludeChild;
+      return this;
+    }
+
     public Builder headerTable(Table<String, String, String> headerTable) {
       this.headerTable = headerTable;
       return this;
@@ -350,7 +369,8 @@ public Builder fileNameHeader(String fileNameHeader) {
     }
 
     public ReliableTaildirEventReader build() throws IOException {
-      return new ReliableTaildirEventReader(filePaths, headerTable, positionFilePath, skipToEnd,
+      return new ReliableTaildirEventReader(filePaths, filePathsIncludeChild, headerTable,
+                                            positionFilePath, skipToEnd,
                                             addByteOffset, cachePatternMatching,
                                             annotateFileName, fileNameHeader);
     }

flume-ng-sources/flume-taildir-source/src/main/java/org/apache/flume/source/taildir/TailMatcher.java

@@ -0,0 +1,34 @@
+/*
+ * 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.source.taildir;
+
+import java.io.File;
+import java.util.List;
+
+/**
+ * Identifies and caches the files matched
+ */
+public interface TailMatcher {
+  List<File> getMatchingFiles();
+
+  String getFileGroup();
+
+  void deleteFileCache(File file);
+}