STORM-2373: HDFS Spout should support multiple ignore extensions

external/storm-hdfs/README.md

@@ -466,7 +466,8 @@ visible to the spout. This can be achieved by either writing the files out to an
 and once completely written, move it to the monitored directory. Alternatively the file
 can be created with a '.ignore' suffix in the monitored directory and after data is completely 
 written, rename it without the suffix. File names with a '.ignore' suffix are ignored
-by the spout.
+by the spout. However you can also set your own comma separated suffixes to ignore certain files 
+using api ```setIgnoreSuffix (String)```
 
 When the spout is actively consuming a file, it renames the file with a '.inprogress' suffix.
 After consuming all the contents in the file, the file will be moved to a configurable *done* 
@@ -547,7 +548,7 @@ Only methods mentioned in **bold** are required.
 | **.setArchiveDir()**       |~~hdfsspout.archive.dir~~             |             | After a file is processed completely it will be moved to this HDFS directory. If this directory does not exist it will be created. E.g. /data/done|
 | **.setBadFilesDir()**      |~~hdfsspout.badfiles.dir~~            |             | if there is an error parsing a file's contents, the file is moved to this location.  If this directory does not exist it will be created. E.g. /data/badfiles  |
 | .setLockDir()              |~~hdfsspout.lock.dir~~                | '.lock' subdirectory under hdfsspout.source.dir | Dir in which lock files will be created. Concurrent HDFS spout instances synchronize using *lock* files. Before processing a file the spout instance creates a lock file in this directory with same name as input file and deletes this lock file after processing the file. Spouts also periodically makes a note of their progress (wrt reading the input file) in the lock file so that another spout instance can resume progress on the same file if the spout dies for any reason.|
-| .setIgnoreSuffix()         |~~hdfsspout.ignore.suffix~~           |   .ignore   | File names with this suffix in the in the hdfsspout.source.dir location will not be processed|
+| .setIgnoreSuffix()         |~~hdfsspout.ignore.suffix~~           |   .ignore   | Comma separated list of file name suffixes. Files with matching suffixes in the hdfsspout.source.dir location will not be processed |
 | .setCommitFrequencyCount() |~~hdfsspout.commit.count~~            |    20000    | Record progress in the lock file after these many records are processed. If set to 0, this criterion will not be used. |
 | .setCommitFrequencySec()   |~~hdfsspout.commit.sec~~              |    10       | Record progress in the lock file after these many seconds have elapsed. Must be greater than 0 |
 | .setMaxOutstanding()       |~~hdfsspout.max.outstanding~~         |   10000     | Limits the number of unACKed tuples by pausing tuple generation (if ACKers are used in the topology) |

external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/HdfsUtils.java

@@ -18,20 +18,20 @@
 
 package org.apache.storm.hdfs.common;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.protocol.AlreadyBeingCreatedException;
 import org.apache.hadoop.ipc.RemoteException;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-
 public class HdfsUtils {
   /** list files sorted by modification time that have not been modified since 'olderThan'. if
    * 'olderThan' is <= 0 then the filtering is disabled */
@@ -76,6 +76,49 @@ public static FSDataOutputStream tryCreateFile(FileSystem fs, Path file) throws
       }
     }
   }
+  /**
+   * list files sorted by modification time that have not been modified since
+   * 'olderThan'. if 'olderThan' is <= 0 then the filtering is disabled
+   *
+   * @param fs
+   *            - {@link FileSystem}
+   * @param directory
+   *            - Directory in which it will look for
+   * @param olderThan
+   *            - Files updated olderthan this time
+   * @param ignoreSuffixes
+   *            - List of ignoreSuffixes
+   * @return - List file path satisfied by criteria
+   * @throws IOException
+   */
+  public static ArrayList<Path> listFilesByModificationTimeWithIgnoreSuffixes(FileSystem fs, Path directory,
+      long olderThan, List<String> ignoreSuffixes) throws IOException {
+    ArrayList<Path> list = listFilesByModificationTime(fs, directory, olderThan);
+    ArrayList<Path> result = new ArrayList<>(list.size());
+    for (Path path : list) {
+      if (!filterSufix(path.getName(), ignoreSuffixes)) {
+        result.add(path);
+      }
+    }
+    return result;
+  }
+
+  /**
+   *
+   * @param name
+   *            -name of file
+   * @param ignoreSuffixes
+   *            - List of suffixes to be ignored
+   * @return
+   */
+  private static boolean filterSufix(String name, List<String> ignoreSuffixes) {
+    for (String suffix : ignoreSuffixes) {
+      if (name.endsWith(suffix)) {
+        return true;
+      }
+    }
+    return false;
+  }
 
   public static class Pair<K,V> {
     private K key;

external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/FileOffset.java

@@ -29,7 +29,7 @@
  *       - constructor(string) for deserialization
  */
 
-interface FileOffset extends Comparable<FileOffset>, Cloneable {
+public interface FileOffset extends Comparable<FileOffset>, Cloneable {
   /** tests if rhs == currOffset+1 */
   boolean isNextOffset(FileOffset rhs);
   FileOffset clone();

external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java

@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -30,20 +32,20 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import org.apache.storm.Config;
+import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.storm.Config;
 import org.apache.storm.hdfs.common.HdfsUtils;
 import org.apache.storm.hdfs.common.security.HdfsSecurityUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import org.apache.storm.spout.SpoutOutputCollector;
 import org.apache.storm.task.TopologyContext;
 import org.apache.storm.topology.OutputFieldsDeclarer;
 import org.apache.storm.topology.base.BaseRichSpout;
 import org.apache.storm.tuple.Fields;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class HdfsSpout extends BaseRichSpout {
 
@@ -71,7 +73,7 @@ public class HdfsSpout extends BaseRichSpout {
   private boolean clocksInSync = true;
 
   private String inprogress_suffix = ".inprogress"; // not configurable to prevent change between topology restarts
-  private String ignoreSuffix = ".ignore";
+  private List<String> ignoreSuffixes = new ArrayList<>();
 
   private String outputStreamName= null;
 
@@ -164,11 +166,20 @@ public HdfsSpout setClocksInSync(boolean clocksInSync) {
   }
 
 
-  public HdfsSpout setIgnoreSuffix(String ignoreSuffix) {
-    this.ignoreSuffix = ignoreSuffix;
+  public HdfsSpout setIgnoreSuffix(String ignoreSuffix) { 
+	String[] suffixes = ignoreSuffix.toString().split(",");
+	if (suffixes!=null) {
+		for (String suffix : suffixes) {
+			String trimmedSuffix = StringUtils.trim(suffix);
+			if (StringUtils.isNotEmpty(trimmedSuffix)) {
+				ignoreSuffixes.add(trimmedSuffix);
+		}
+	  }
+  }
     return this;
   }
 
+
   /** Output field names. Number of fields depends upon the reader type */
   public HdfsSpout withOutputFields(String... fields) {
     outputFields = new Fields(fields);
@@ -447,7 +458,20 @@ public void open(Map conf, TopologyContext context, SpoutOutputCollector collect
 
     // -- ignore file names config
     if ( conf.containsKey(Configs.IGNORE_SUFFIX) ) {
-      this.ignoreSuffix = conf.get(Configs.IGNORE_SUFFIX).toString();
+      String[] suffixes = conf.get(Configs.IGNORE_SUFFIX).toString().split(",");
+      if (suffixes!=null) {
+    	  for (String suffix : suffixes) {
+			String trimmedSuffix = StringUtils.trim(suffix);
+			if (StringUtils.isNotEmpty(trimmedSuffix)) {
+				ignoreSuffixes.add(trimmedSuffix);
+			}
+    	  }
+      }
+    }
+
+    //To support backward compatibility
+    if (ignoreSuffixes.isEmpty()) {
+      ignoreSuffixes.add(".ignore");
     }
 
     // -- lock dir config
@@ -596,15 +620,13 @@ private FileReader pickNextFile() {
       }
 
       // 2) If no abandoned files, then pick oldest file in sourceDirPath, lock it and rename it
-      Collection<Path> listing = HdfsUtils.listFilesByModificationTime(hdfs, sourceDirPath, 0);
+	  Collection<Path> listing = HdfsUtils.listFilesByModificationTimeWithIgnoreSuffixes(hdfs, sourceDirPath,0,ignoreSuffixes);
 
       for (Path file : listing) {
         if (file.getName().endsWith(inprogress_suffix)) {
           continue;
         }
-        if (file.getName().endsWith(ignoreSuffix)) {
-          continue;
-        }
+
         lock = FileLock.tryLock(hdfs, file, lockDirPath, spoutId);
         if (lock == null) {
           LOG.debug("Unable to get FileLock for {}, so skipping it.", file);