diff --git a/external/storm-hdfs/README.md b/external/storm-hdfs/README.md index c7ab7ca4bde..beaaf1acef1 100644 --- a/external/storm-hdfs/README.md +++ b/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) | diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/HdfsUtils.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/HdfsUtils.java index 5ec533359d7..0a56a1a58f5 100644 --- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/common/HdfsUtils.java +++ b/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 listFilesByModificationTimeWithIgnoreSuffixes(FileSystem fs, Path directory, + long olderThan, List ignoreSuffixes) throws IOException { + ArrayList list = listFilesByModificationTime(fs, directory, olderThan); + ArrayList 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 ignoreSuffixes) { + for (String suffix : ignoreSuffixes) { + if (name.endsWith(suffix)) { + return true; + } + } + return false; + } public static class Pair { private K key; diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/FileOffset.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/FileOffset.java index ad487796239..1b5a23e7b5f 100644 --- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/FileOffset.java +++ b/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, Cloneable { +public interface FileOffset extends Comparable, Cloneable { /** tests if rhs == currOffset+1 */ boolean isNextOffset(FileOffset rhs); FileOffset clone(); diff --git a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java b/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java index b7627f24178..8903e7eea15 100644 --- a/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/spout/HdfsSpout.java +++ b/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 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 listing = HdfsUtils.listFilesByModificationTime(hdfs, sourceDirPath, 0); + Collection 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); diff --git a/external/storm-hdfs/src/test/java/org/apache/storm/hdfs/spout/TestHdfsSpout.java b/external/storm-hdfs/src/test/java/org/apache/storm/hdfs/spout/TestHdfsSpout.java index f60cbf3e315..05a1e1757a0 100644 --- a/external/storm-hdfs/src/test/java/org/apache/storm/hdfs/spout/TestHdfsSpout.java +++ b/external/storm-hdfs/src/test/java/org/apache/storm/hdfs/spout/TestHdfsSpout.java @@ -18,16 +18,19 @@ package org.apache.storm.hdfs.spout; -import org.apache.storm.Config; -import org.apache.storm.spout.SpoutOutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.hadoop.hdfs.DistributedFileSystem; -import org.apache.hadoop.hdfs.MiniDFSCluster; -import org.apache.hadoop.io.Writable; -import org.apache.hadoop.util.ReflectionUtils; -import org.apache.storm.hdfs.common.HdfsUtils; -import org.junit.AfterClass; -import org.junit.Assert; +import java.io.BufferedReader; +import java.io.File; +import java.io.IOException; +import java.io.InputStreamReader; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.NavigableSet; +import java.util.TreeSet; + import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FSDataOutputStream; @@ -35,725 +38,771 @@ 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.MiniDFSCluster; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.SequenceFile; import org.apache.hadoop.io.Text; -import org.junit.Before; +import org.apache.hadoop.io.Writable; +import org.apache.hadoop.util.ReflectionUtils; +import org.apache.storm.Config; +import org.apache.storm.hdfs.common.HdfsUtils; +import org.apache.storm.hdfs.common.HdfsUtils.Pair; +import org.apache.storm.spout.SpoutOutputCollector; +import org.apache.storm.task.TopologyContext; import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; -import java.io.BufferedReader; -import java.io.File; -import java.io.IOException; -import java.io.InputStreamReader; -import java.lang.reflect.Field; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.storm.hdfs.common.HdfsUtils.Pair; - public class TestHdfsSpout { - @Rule - public TemporaryFolder tempFolder = new TemporaryFolder(); - public File baseFolder; - private Path source; - private Path archive; - private Path badfiles; - - - public TestHdfsSpout() { - } - - static MiniDFSCluster.Builder builder; - static MiniDFSCluster hdfsCluster; - static DistributedFileSystem fs; - static String hdfsURI; - static Configuration conf = new Configuration(); - - @BeforeClass - public static void setupClass() throws IOException { - builder = new MiniDFSCluster.Builder(new Configuration()); - hdfsCluster = builder.build(); - fs = hdfsCluster.getFileSystem(); - hdfsURI = "hdfs://localhost:" + hdfsCluster.getNameNodePort() + "/"; - } - - @AfterClass - public static void teardownClass() throws IOException { - fs.close(); - hdfsCluster.shutdown(); - } - - - @Before - public void setup() throws Exception { - baseFolder = tempFolder.newFolder("hdfsspout"); - source = new Path(baseFolder.toString() + "/source"); - fs.mkdirs(source); - archive = new Path(baseFolder.toString() + "/archive"); - fs.mkdirs(archive); - badfiles = new Path(baseFolder.toString() + "/bad"); - fs.mkdirs(badfiles); - } - - @After - public void shutDown() throws IOException { - fs.delete(new Path(baseFolder.toString()), true); - } - - @Test - public void testSimpleText_noACK() throws IOException { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 5); - - Path file2 = new Path(source.toString() + "/file2.txt"); - createTextFile(file2, 5); - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1); - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - runSpout(spout,"r11"); - - Path arc1 = new Path(archive.toString() + "/file1.txt"); - Path arc2 = new Path(archive.toString() + "/file2.txt"); - checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1, arc2); - } - - @Test - public void testSimpleText_ACK() throws IOException { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 5); - - Path file2 = new Path(source.toString() + "/file2.txt"); - createTextFile(file2, 5); - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1); - - Map conf = getCommonConfigs(); - conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "1"); // enable ACKing - openSpout(spout, 0, conf); - - // consume file 1 - runSpout(spout, "r6", "a0", "a1", "a2", "a3", "a4"); - Path arc1 = new Path(archive.toString() + "/file1.txt"); - checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1); - - // consume file 2 - runSpout(spout, "r6", "a5", "a6", "a7", "a8", "a9"); - Path arc2 = new Path(archive.toString() + "/file2.txt"); - checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1, arc2); - } - - @Test - public void testResumeAbandoned_Text_NoAck() throws Exception { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 6); - - final Integer lockExpirySec = 1; - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1000); // effectively disable commits based on time - spout.setLockTimeoutSec(lockExpirySec); - - - HdfsSpout spout2 = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout2.setCommitFrequencyCount(1); - spout2.setCommitFrequencySec(1000); // effectively disable commits based on time - spout2.setLockTimeoutSec(lockExpirySec); - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - openSpout(spout2, 1, conf); - - // consume file 1 partially - List res = runSpout(spout, "r2"); - Assert.assertEquals(2, res.size()); - - // abandon file - FileLock lock = getField(spout, "lock"); - TestFileLock.closeUnderlyingLockFile(lock); - Thread.sleep(lockExpirySec * 2 * 1000); - - // check lock file presence - Assert.assertTrue(fs.exists(lock.getLockFile())); - - // create another spout to take over processing and read a few lines - List res2 = runSpout(spout2, "r3"); - Assert.assertEquals(3, res2.size()); - - // check lock file presence - Assert.assertTrue(fs.exists(lock.getLockFile())); - - // check lock file contents - List contents = readTextFile(fs, lock.getLockFile().toString()); - Assert.assertFalse(contents.isEmpty()); - - // finish up reading the file - res2 = runSpout(spout2, "r2"); - Assert.assertEquals(4, res2.size()); - - // check lock file is gone - Assert.assertFalse(fs.exists(lock.getLockFile())); - FileReader rdr = getField(spout2, "reader"); - Assert.assertNull(rdr); - Assert.assertTrue(getBoolField(spout2, "fileReadCompletely")); - - } - - @Test - public void testResumeAbandoned_Seq_NoAck() throws Exception { - Path file1 = new Path(source.toString() + "/file1.seq"); - createSeqFile(fs, file1, 6); - - final Integer lockExpirySec = 1; - - HdfsSpout spout = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1000); // effectively disable commits based on time - spout.setLockTimeoutSec(lockExpirySec); - - - HdfsSpout spout2 = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); - spout2.setCommitFrequencyCount(1); - spout2.setCommitFrequencySec(1000); // effectively disable commits based on time - spout2.setLockTimeoutSec(lockExpirySec); - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - openSpout(spout2, 1, conf); - - // consume file 1 partially - List res = runSpout(spout, "r2"); - Assert.assertEquals(2, res.size()); - // abandon file - FileLock lock = getField(spout, "lock"); - TestFileLock.closeUnderlyingLockFile(lock); - Thread.sleep(lockExpirySec * 2 * 1000); - - // check lock file presence - Assert.assertTrue(fs.exists(lock.getLockFile())); - - // create another spout to take over processing and read a few lines - List res2 = runSpout(spout2, "r3"); - Assert.assertEquals(3, res2.size()); - - // check lock file presence - Assert.assertTrue(fs.exists(lock.getLockFile())); - - // check lock file contents - List contents = getTextFileContents(fs, lock.getLockFile()); - Assert.assertFalse(contents.isEmpty()); - - // finish up reading the file - res2 = runSpout(spout2, "r3"); - Assert.assertEquals(4, res2.size()); - - // check lock file is gone - Assert.assertFalse(fs.exists(lock.getLockFile())); - FileReader rdr = getField(spout2, "reader"); - Assert.assertNull( rdr ); - Assert.assertTrue(getBoolField(spout2, "fileReadCompletely")); - } - - private void checkCollectorOutput_txt(MockCollector collector, Path... txtFiles) throws IOException { - ArrayList expected = new ArrayList<>(); - for (Path txtFile : txtFiles) { - List lines= getTextFileContents(fs, txtFile); - expected.addAll(lines); - } - - List actual = new ArrayList<>(); - for (Pair> item : collector.items) { - actual.add(item.getValue().get(0).toString()); - } - Assert.assertEquals(expected, actual); - } - - private List getTextFileContents(FileSystem fs, Path txtFile) throws IOException { - ArrayList result = new ArrayList<>(); - FSDataInputStream istream = fs.open(txtFile); - InputStreamReader isreader = new InputStreamReader(istream,"UTF-8"); - BufferedReader reader = new BufferedReader(isreader); - - for( String line = reader.readLine(); line!=null; line = reader.readLine() ) { - result.add(line); - } - isreader.close(); - return result; - } - - - private void checkCollectorOutput_seq(MockCollector collector, Path... seqFiles) throws IOException { - ArrayList expected = new ArrayList<>(); - for (Path seqFile : seqFiles) { - List lines= getSeqFileContents(fs, seqFile); - expected.addAll(lines); - } - Assert.assertTrue(expected.equals(collector.lines)); - } - - private List getSeqFileContents(FileSystem fs, Path... seqFiles) throws IOException { - ArrayList result = new ArrayList<>(); - - for (Path seqFile : seqFiles) { - Path file = new Path(fs.getUri().toString() + seqFile.toString()); - SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file)); - try { - Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf); - Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf); - while (reader.next(key, value)) { - String keyValStr = Arrays.asList(key, value).toString(); - result.add(keyValStr); - } - } finally { - reader.close(); - } - }// for - return result; - } - - private List listDir(Path p) throws IOException { - ArrayList result = new ArrayList<>(); - RemoteIterator fileNames = fs.listFiles(p, false); - while ( fileNames.hasNext() ) { - LocatedFileStatus fileStatus = fileNames.next(); - result.add(Path.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toString()); - } - return result; - } - - - @Test - public void testMultipleFileConsumption_Ack() throws Exception { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 5); - - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1); - - Map conf = getCommonConfigs(); - conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "1"); // enable ACKing - openSpout(spout, 0, conf); - - // read few lines from file1 dont ack - runSpout(spout, "r3"); - FileReader reader = getField(spout, "reader"); - Assert.assertNotNull(reader); - Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); - - // read remaining lines - runSpout(spout, "r3"); - reader = getField(spout, "reader"); - Assert.assertNotNull(reader); - Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely") ); - - // ack few - runSpout(spout, "a0", "a1", "a2"); - reader = getField(spout, "reader"); - Assert.assertNotNull(reader); - Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); - - //ack rest - runSpout(spout, "a3", "a4"); - reader = getField(spout, "reader"); - Assert.assertNull(reader); - Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); - - - // go to next file - Path file2 = new Path(source.toString() + "/file2.txt"); - createTextFile(file2, 5); - - // Read 1 line - runSpout(spout, "r1"); - Assert.assertNotNull(getField(spout, "reader")); - Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); - - // ack 1 tuple - runSpout(spout, "a5"); - Assert.assertNotNull(getField(spout, "reader")); - Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); - - - // read and ack remaining lines - runSpout(spout, "r5", "a6", "a7", "a8", "a9"); - Assert.assertNull(getField(spout, "reader")); - Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); - } - - private static T getField(HdfsSpout spout, String fieldName) throws NoSuchFieldException, IllegalAccessException { - Field readerFld = HdfsSpout.class.getDeclaredField(fieldName); - readerFld.setAccessible(true); - return (T) readerFld.get(spout); - } - - private static boolean getBoolField(HdfsSpout spout, String fieldName) throws NoSuchFieldException, IllegalAccessException { - Field readerFld = HdfsSpout.class.getDeclaredField(fieldName); - readerFld.setAccessible(true); - return readerFld.getBoolean(spout); - } - - - @Test - public void testSimpleSequenceFile() throws IOException { - //1) create a couple files to consume - source = new Path("/tmp/hdfsspout/source"); - fs.mkdirs(source); - archive = new Path("/tmp/hdfsspout/archive"); - fs.mkdirs(archive); - - Path file1 = new Path(source + "/file1.seq"); - createSeqFile(fs, file1, 5); - - Path file2 = new Path(source + "/file2.seq"); - createSeqFile(fs, file2, 5); - - - HdfsSpout spout = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - // consume both files - List res = runSpout(spout, "r11"); - Assert.assertEquals(10, res.size()); - - Assert.assertEquals(2, listDir(archive).size()); - - - Path f1 = new Path(archive + "/file1.seq"); - Path f2 = new Path(archive + "/file2.seq"); - - checkCollectorOutput_seq((MockCollector) spout.getCollector(), f1, f2); - } - - @Test - public void testReadFailures() throws Exception { - // 1) create couple of input files to read - Path file1 = new Path(source.toString() + "/file1.txt"); - Path file2 = new Path(source.toString() + "/file2.txt"); - - createTextFile(file1, 6); - createTextFile(file2, 7); - Assert.assertEquals(2, listDir(source).size()); - - // 2) run spout - HdfsSpout spout = makeSpout(MockTextFailingReader.class.getName(), MockTextFailingReader.defaultFields); - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - List res = runSpout(spout, "r11"); - String[] expected = new String[] {"[line 0]","[line 1]","[line 2]","[line 0]","[line 1]","[line 2]"}; - Assert.assertArrayEquals(expected, res.toArray()); - - // 3) make sure 6 lines (3 from each file) were read in all - Assert.assertEquals(((MockCollector) spout.getCollector()).lines.size(), 6); - ArrayList badFiles = HdfsUtils.listFilesByModificationTime(fs, badfiles, 0); - Assert.assertEquals(badFiles.size(), 2); - } - - // check lock creation/deletion and contents - @Test - public void testLocking() throws Exception { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 10); - - // 0) config spout to log progress in lock file for each tuple - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(1); - spout.setCommitFrequencySec(1000); // effectively disable commits based on time - - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - // 1) read initial lines in file, then check if lock exists - List res = runSpout(spout, "r5"); - Assert.assertEquals(5, res.size()); - List lockFiles = listDir(spout.getLockDirPath()); - Assert.assertEquals(1, lockFiles.size()); - - // 2) check log file content line count == tuples emitted + 1 - List lines = readTextFile(fs, lockFiles.get(0)); - Assert.assertEquals(lines.size(), res.size()+1); - - // 3) read remaining lines in file, then ensure lock is gone - runSpout(spout, "r6"); - lockFiles = listDir(spout.getLockDirPath()); - Assert.assertEquals(0, lockFiles.size()); - - - // 4) --- Create another input file and reverify same behavior --- - Path file2 = new Path(source.toString() + "/file2.txt"); - createTextFile(file2, 10); - - // 5) read initial lines in file, then check if lock exists - res = runSpout(spout, "r5"); - Assert.assertEquals(15, res.size()); - lockFiles = listDir(spout.getLockDirPath()); - Assert.assertEquals(1, lockFiles.size()); - - // 6) check log file content line count == tuples emitted + 1 - lines = readTextFile(fs, lockFiles.get(0)); - Assert.assertEquals(6, lines.size()); - - // 7) read remaining lines in file, then ensure lock is gone - runSpout(spout, "r6"); - lockFiles = listDir(spout.getLockDirPath()); - Assert.assertEquals(0, lockFiles.size()); - } - - @Test - public void testLockLoggingFreqCount() throws Exception { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 10); - - // 0) config spout to log progress in lock file for each tuple - - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(2); // 1 lock log entry every 2 tuples - spout.setCommitFrequencySec(1000); // Effectively disable commits based on time - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - // 1) read 5 lines in file, - runSpout(spout, "r5"); - - // 2) check log file contents - String lockFile = listDir(spout.getLockDirPath()).get(0); - List lines = readTextFile(fs, lockFile); - Assert.assertEquals(lines.size(), 3); - - // 3) read 6th line and see if another log entry was made - runSpout(spout, "r1"); - lines = readTextFile(fs, lockFile); - Assert.assertEquals(lines.size(), 4); - } - - @Test - public void testLockLoggingFreqSec() throws Exception { - Path file1 = new Path(source.toString() + "/file1.txt"); - createTextFile(file1, 10); - - // 0) config spout to log progress in lock file for each tuple - HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); - spout.setCommitFrequencyCount(0); // disable it - spout.setCommitFrequencySec(2); // log every 2 sec - - Map conf = getCommonConfigs(); - openSpout(spout, 0, conf); - - // 1) read 5 lines in file - runSpout(spout, "r5"); - - // 2) check log file contents - String lockFile = listDir(spout.getLockDirPath()).get(0); - List lines = readTextFile(fs, lockFile); - Assert.assertEquals(lines.size(), 1); - Thread.sleep(3000); // allow freq_sec to expire - - // 3) read another line and see if another log entry was made - runSpout(spout, "r1"); - lines = readTextFile(fs, lockFile); - Assert.assertEquals(2, lines.size()); - } - - private static List readTextFile(FileSystem fs, String f) throws IOException { - Path file = new Path(f); - FSDataInputStream x = fs.open(file); - BufferedReader reader = new BufferedReader(new InputStreamReader(x)); - String line = null; - ArrayList result = new ArrayList<>(); - while( (line = reader.readLine()) !=null ) - result.add( line ); - return result; - } - - - private Map getCommonConfigs() { - Map conf = new HashMap(); - conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "0"); - return conf; - } - - private HdfsSpout makeSpout(String readerType, String[] outputFields) { - HdfsSpout spout = new HdfsSpout().withOutputFields(outputFields) - .setReaderType(readerType) - .setHdfsUri(hdfsCluster.getURI().toString()) - .setSourceDir(source.toString()) - .setArchiveDir(archive.toString()) - .setBadFilesDir(badfiles.toString()); - - return spout; - } - - private void openSpout(HdfsSpout spout, int spoutId, Map conf) { - MockCollector collector = new MockCollector(); - spout.open(conf, new MockTopologyContext(spoutId), collector); - } - - /** - * Execute a sequence of calls on HdfsSpout. - * - * @param cmds: set of commands to run, - * e.g. "r,r,r,r,a1,f2,...". The commands are: - * r[N] - receive() called N times - * aN - ack, item number: N - * fN - fail, item number: N - */ - - private List runSpout(HdfsSpout spout, String... cmds) { - MockCollector collector = (MockCollector) spout.getCollector(); - for(String cmd : cmds) { - if(cmd.startsWith("r")) { - int count = 1; - if(cmd.length() > 1) { - count = Integer.parseInt(cmd.substring(1)); - } - for(int i=0; i> item = collector.items.get(n); - spout.ack(item.getKey()); - } - else if(cmd.startsWith("f")) { - int n = Integer.parseInt(cmd.substring(1)); - Pair> item = collector.items.get(n); - spout.fail(item.getKey()); - } - } - return collector.lines; - } - - private void createTextFile(Path file, int lineCount) throws IOException { - FSDataOutputStream os = fs.create(file); - int size = 0; - for (int i = 0; i < lineCount; i++) { - os.writeBytes("line " + i + System.lineSeparator()); - String msg = "line " + i + System.lineSeparator(); - size += msg.getBytes().length; - } - os.close(); - } - - - - private static void createSeqFile(FileSystem fs, Path file, int rowCount) throws IOException { - - Configuration conf = new Configuration(); - try { - if(fs.exists(file)) { - fs.delete(file, false); - } - - SequenceFile.Writer w = SequenceFile.createWriter(fs, conf, file, IntWritable.class, Text.class ); - for (int i = 0; i < rowCount; i++) { - w.append(new IntWritable(i), new Text("line " + i)); - } - w.close(); - System.out.println("done"); - } catch (IOException e) { - e.printStackTrace(); - - } - } - - - - static class MockCollector extends SpoutOutputCollector { - //comma separated offsets - public ArrayList lines; - public ArrayList > > items; - - public MockCollector() { - super(null); - lines = new ArrayList<>(); - items = new ArrayList<>(); - } - - - - @Override - public List emit(String streamId, List tuple, Object messageId) { - lines.add(tuple.toString()); - items.add(HdfsUtils.Pair.of(messageId, tuple)); - return null; - } - - @Override - public void emitDirect(int arg0, String arg1, List arg2, Object arg3) { - throw new UnsupportedOperationException("NOT Implemented"); - } - - @Override - public void reportError(Throwable arg0) { - throw new UnsupportedOperationException("NOT Implemented"); - } - - @Override - public long getPendingCount() { - return 0; - } - } // class MockCollector - - - - // Throws IOExceptions for 3rd & 4th call to next(), succeeds on 5th, thereafter - // throws ParseException. Effectively produces 3 lines (1,2 & 3) from each file read - static class MockTextFailingReader extends TextFileReader { - public static final String[] defaultFields = {"line"}; - int readAttempts = 0; - - public MockTextFailingReader(FileSystem fs, Path file, Map conf) throws IOException { - super(fs, file, conf); - } - - @Override - public List next() throws IOException, ParseException { - readAttempts++; - if (readAttempts == 3 || readAttempts ==4) { - throw new IOException("mock test exception"); - } else if (readAttempts > 5 ) { - throw new ParseException("mock test exception", null); - } - return super.next(); - } - } - - static class MockTopologyContext extends TopologyContext { - private final int componentId; - - public MockTopologyContext(int componentId) { - // StormTopology topology, Map stormConf, Map taskToComponent, Map> componentToSortedTasks, Map> componentToStreamToFields, String stormId, String codeDir, String pidDir, Integer taskId, Integer workerPort, List workerTasks, Map defaultResources, Map userResources, Map executorData, Map>> registeredMetrics, Atom openOrPrepareWasCalled - super(null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null); - this.componentId = componentId; - } - - public String getThisComponentId() { - return Integer.toString( componentId ); - } + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + public File baseFolder; + private Path source; + private Path archive; + private Path badfiles; + + + public TestHdfsSpout() { + } + + static MiniDFSCluster.Builder builder; + static MiniDFSCluster hdfsCluster; + static DistributedFileSystem fs; + static String hdfsURI; + static Configuration conf = new Configuration(); + + @BeforeClass + public static void setupClass() throws IOException { + builder = new MiniDFSCluster.Builder(new Configuration()); + hdfsCluster = builder.build(); + fs = hdfsCluster.getFileSystem(); + hdfsURI = "hdfs://localhost:" + hdfsCluster.getNameNodePort() + "/"; + } + + @AfterClass + public static void teardownClass() throws IOException { + fs.close(); + hdfsCluster.shutdown(); + } + + + @Before + public void setup() throws Exception { + baseFolder = tempFolder.newFolder("hdfsspout"); + source = new Path(baseFolder.toString() + "/source"); + fs.mkdirs(source); + archive = new Path(baseFolder.toString() + "/archive"); + fs.mkdirs(archive); + badfiles = new Path(baseFolder.toString() + "/bad"); + fs.mkdirs(badfiles); + } + + @After + public void shutDown() throws IOException { + fs.delete(new Path(baseFolder.toString()), true); + } + + @Test + public void testSimpleText_noACK() throws IOException { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 5); + + Path file2 = new Path(source.toString() + "/file2.txt"); + createTextFile(file2, 5); + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1); + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + runSpout(spout,"r11"); + + Path arc1 = new Path(archive.toString() + "/file1.txt"); + Path arc2 = new Path(archive.toString() + "/file2.txt"); + checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1, arc2); + } + + @Test + public void testSimpleText_ACK() throws IOException { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 5); + + Path file2 = new Path(source.toString() + "/file2.txt"); + createTextFile(file2, 5); + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1); + + Map conf = getCommonConfigs(); + conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "1"); // enable ACKing + openSpout(spout, 0, conf); + + // consume file 1 + runSpout(spout, "r6", "a0", "a1", "a2", "a3", "a4"); + Path arc1 = new Path(archive.toString() + "/file1.txt"); + checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1); + + // consume file 2 + runSpout(spout, "r6", "a5", "a6", "a7", "a8", "a9"); + Path arc2 = new Path(archive.toString() + "/file2.txt"); + checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1, arc2); + } + + @Test + public void testResumeAbandoned_Text_NoAck() throws Exception { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 6); + + final Integer lockExpirySec = 1; + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1000); // effectively disable commits based on time + spout.setLockTimeoutSec(lockExpirySec); + + + HdfsSpout spout2 = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout2.setCommitFrequencyCount(1); + spout2.setCommitFrequencySec(1000); // effectively disable commits based on time + spout2.setLockTimeoutSec(lockExpirySec); + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + openSpout(spout2, 1, conf); + + // consume file 1 partially + List res = runSpout(spout, "r2"); + Assert.assertEquals(2, res.size()); + + // abandon file + FileLock lock = getField(spout, "lock"); + TestFileLock.closeUnderlyingLockFile(lock); + Thread.sleep(lockExpirySec * 2 * 1000); + + // check lock file presence + Assert.assertTrue(fs.exists(lock.getLockFile())); + + // create another spout to take over processing and read a few lines + List res2 = runSpout(spout2, "r3"); + Assert.assertEquals(3, res2.size()); + + // check lock file presence + Assert.assertTrue(fs.exists(lock.getLockFile())); + + // check lock file contents + List contents = readTextFile(fs, lock.getLockFile().toString()); + Assert.assertFalse(contents.isEmpty()); + + // finish up reading the file + res2 = runSpout(spout2, "r2"); + Assert.assertEquals(4, res2.size()); + + // check lock file is gone + Assert.assertFalse(fs.exists(lock.getLockFile())); + FileReader rdr = getField(spout2, "reader"); + Assert.assertNull(rdr); + Assert.assertTrue(getBoolField(spout2, "fileReadCompletely")); + + } + + @Test + public void testResumeAbandoned_Seq_NoAck() throws Exception { + Path file1 = new Path(source.toString() + "/file1.seq"); + createSeqFile(fs, file1, 6); + + final Integer lockExpirySec = 1; + + HdfsSpout spout = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1000); // effectively disable commits based on time + spout.setLockTimeoutSec(lockExpirySec); + + + HdfsSpout spout2 = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); + spout2.setCommitFrequencyCount(1); + spout2.setCommitFrequencySec(1000); // effectively disable commits based on time + spout2.setLockTimeoutSec(lockExpirySec); + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + openSpout(spout2, 1, conf); + + // consume file 1 partially + List res = runSpout(spout, "r2"); + Assert.assertEquals(2, res.size()); + // abandon file + FileLock lock = getField(spout, "lock"); + TestFileLock.closeUnderlyingLockFile(lock); + Thread.sleep(lockExpirySec * 2 * 1000); + + // check lock file presence + Assert.assertTrue(fs.exists(lock.getLockFile())); + + // create another spout to take over processing and read a few lines + List res2 = runSpout(spout2, "r3"); + Assert.assertEquals(3, res2.size()); + + // check lock file presence + Assert.assertTrue(fs.exists(lock.getLockFile())); + + // check lock file contents + List contents = getTextFileContents(fs, lock.getLockFile()); + Assert.assertFalse(contents.isEmpty()); + + // finish up reading the file + res2 = runSpout(spout2, "r3"); + Assert.assertEquals(4, res2.size()); + + // check lock file is gone + Assert.assertFalse(fs.exists(lock.getLockFile())); + FileReader rdr = getField(spout2, "reader"); + Assert.assertNull( rdr ); + Assert.assertTrue(getBoolField(spout2, "fileReadCompletely")); + } + + private void checkCollectorOutput_txt(MockCollector collector, Path... txtFiles) throws IOException { + ArrayList expected = new ArrayList<>(); + for (Path txtFile : txtFiles) { + List lines= getTextFileContents(fs, txtFile); + expected.addAll(lines); + } + + List actual = new ArrayList<>(); + for (Pair> item : collector.items) { + actual.add(item.getValue().get(0).toString()); + } + Assert.assertEquals(expected, actual); + } + + private List getTextFileContents(FileSystem fs, Path txtFile) throws IOException { + ArrayList result = new ArrayList<>(); + FSDataInputStream istream = fs.open(txtFile); + InputStreamReader isreader = new InputStreamReader(istream,"UTF-8"); + BufferedReader reader = new BufferedReader(isreader); + + for( String line = reader.readLine(); line!=null; line = reader.readLine() ) { + result.add(line); + } + isreader.close(); + return result; + } + + + private void checkCollectorOutput_seq(MockCollector collector, Path... seqFiles) throws IOException { + ArrayList expected = new ArrayList<>(); + for (Path seqFile : seqFiles) { + List lines= getSeqFileContents(fs, seqFile); + expected.addAll(lines); + } + Assert.assertTrue(expected.equals(collector.lines)); + } + + private List getSeqFileContents(FileSystem fs, Path... seqFiles) throws IOException { + ArrayList result = new ArrayList<>(); + + for (Path seqFile : seqFiles) { + Path file = new Path(fs.getUri().toString() + seqFile.toString()); + SequenceFile.Reader reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(file)); + try { + Writable key = (Writable) ReflectionUtils.newInstance(reader.getKeyClass(), conf); + Writable value = (Writable) ReflectionUtils.newInstance(reader.getValueClass(), conf); + while (reader.next(key, value)) { + String keyValStr = Arrays.asList(key, value).toString(); + result.add(keyValStr); + } + } finally { + reader.close(); + } + }// for + return result; + } + + private List listDir(Path p) throws IOException { + ArrayList result = new ArrayList<>(); + RemoteIterator fileNames = fs.listFiles(p, false); + while ( fileNames.hasNext() ) { + LocatedFileStatus fileStatus = fileNames.next(); + result.add(Path.getPathWithoutSchemeAndAuthority(fileStatus.getPath()).toString()); + } + return result; + } + + + @Test + public void testMultipleFileConsumption_Ack() throws Exception { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 5); + + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1); + + Map conf = getCommonConfigs(); + conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "1"); // enable ACKing + openSpout(spout, 0, conf); + + // read few lines from file1 dont ack + runSpout(spout, "r3"); + FileReader reader = getField(spout, "reader"); + Assert.assertNotNull(reader); + Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); + + // read remaining lines + runSpout(spout, "r3"); + reader = getField(spout, "reader"); + Assert.assertNotNull(reader); + Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely") ); + + // ack few + runSpout(spout, "a0", "a1", "a2"); + reader = getField(spout, "reader"); + Assert.assertNotNull(reader); + Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); + + //ack rest + runSpout(spout, "a3", "a4"); + reader = getField(spout, "reader"); + Assert.assertNull(reader); + Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); + + + // go to next file + Path file2 = new Path(source.toString() + "/file2.txt"); + createTextFile(file2, 5); + + // Read 1 line + runSpout(spout, "r1"); + Assert.assertNotNull(getField(spout, "reader")); + Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); + + // ack 1 tuple + runSpout(spout, "a5"); + Assert.assertNotNull(getField(spout, "reader")); + Assert.assertEquals(false, getBoolField(spout, "fileReadCompletely")); + + + // read and ack remaining lines + runSpout(spout, "r5", "a6", "a7", "a8", "a9"); + Assert.assertNull(getField(spout, "reader")); + Assert.assertEquals(true, getBoolField(spout, "fileReadCompletely")); + } + + private static T getField(HdfsSpout spout, String fieldName) throws NoSuchFieldException, IllegalAccessException { + Field readerFld = HdfsSpout.class.getDeclaredField(fieldName); + readerFld.setAccessible(true); + return (T) readerFld.get(spout); + } + + private static boolean getBoolField(HdfsSpout spout, String fieldName) throws NoSuchFieldException, IllegalAccessException { + Field readerFld = HdfsSpout.class.getDeclaredField(fieldName); + readerFld.setAccessible(true); + return readerFld.getBoolean(spout); + } + + @Test + public void testIgnoreFilesFile() throws IOException, InterruptedException { + //1) create a couple files to consume + source = new Path("/tmp/hdfsspout/source"); + if(fs.exists(source)){ + fs.delete(source, true); + } + fs.mkdirs(source); + archive = new Path("/tmp/hdfsspout/archive/"); + if(fs.exists(archive)){ + fs.delete(archive, true); + } + fs.mkdirs(archive); + + Path file1 = new Path(source + "/file1.seq"); + createSeqFile(fs, file1, 5); + file1 = new Path(source + "/file1._COPYING_"); + createSeqFile(fs, file1, 5); + file1 = new Path(source + "/file1.ignore"); + createSeqFile(fs, file1, 5); + file1 = new Path(source + "/file2.seq"); + createSeqFile(fs, file1, 5); + + + HdfsSpout spout = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); + spout.setIgnoreSuffix(".ignore, ._COPYING_"); + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + // consume both files + List res = runSpout(spout, "r11"); + Assert.assertEquals(10, res.size()); + + List listDir = listDir(source); + NavigableSet set = new TreeSet<>(listDir); + String firstMatch = set.ceiling(".ignore"); + Assert.assertNotNull(firstMatch); + firstMatch = set.ceiling("._COPYING_"); + Assert.assertNotNull(firstMatch); + + Assert.assertEquals(2, listDir.size()); + Assert.assertEquals(2, listDir(archive).size()); + + Path f1 = new Path(archive + "/file1.seq"); + Path f2 = new Path(archive + "/file2.seq"); + + checkCollectorOutput_seq((MockCollector) spout.getCollector(), f1, f2); + } + + @Test + public void testSimpleSequenceFile() throws IOException { + //1) create a couple files to consume + source = new Path("/tmp/hdfsspout/source"); + fs.mkdirs(source); + archive = new Path("/tmp/hdfsspout/archive"); + fs.mkdirs(archive); + + Path file1 = new Path(source + "/file1.seq"); + createSeqFile(fs, file1, 5); + + Path file2 = new Path(source + "/file2.seq"); + createSeqFile(fs, file2, 5); + + + HdfsSpout spout = makeSpout(Configs.SEQ, SequenceFileReader.defaultFields); + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + // consume both files + List res = runSpout(spout, "r11"); + Assert.assertEquals(10, res.size()); + + Assert.assertEquals(2, listDir(archive).size()); + + + Path f1 = new Path(archive + "/file1.seq"); + Path f2 = new Path(archive + "/file2.seq"); + + checkCollectorOutput_seq((MockCollector) spout.getCollector(), f1, f2); + } + + @Test + public void testReadFailures() throws Exception { + // 1) create couple of input files to read + Path file1 = new Path(source.toString() + "/file1.txt"); + Path file2 = new Path(source.toString() + "/file2.txt"); + + createTextFile(file1, 6); + createTextFile(file2, 7); + Assert.assertEquals(2, listDir(source).size()); + + // 2) run spout + HdfsSpout spout = makeSpout(MockTextFailingReader.class.getName(), MockTextFailingReader.defaultFields); + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + List res = runSpout(spout, "r11"); + String[] expected = new String[] {"[line 0]","[line 1]","[line 2]","[line 0]","[line 1]","[line 2]"}; + Assert.assertArrayEquals(expected, res.toArray()); + + // 3) make sure 6 lines (3 from each file) were read in all + Assert.assertEquals(((MockCollector) spout.getCollector()).lines.size(), 6); + ArrayList badFiles = HdfsUtils.listFilesByModificationTime(fs, badfiles, 0); + Assert.assertEquals(badFiles.size(), 2); + } + + // check lock creation/deletion and contents + @Test + public void testLocking() throws Exception { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 10); + + // 0) config spout to log progress in lock file for each tuple + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(1); + spout.setCommitFrequencySec(1000); // effectively disable commits based on time + + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + // 1) read initial lines in file, then check if lock exists + List res = runSpout(spout, "r5"); + Assert.assertEquals(5, res.size()); + List lockFiles = listDir(spout.getLockDirPath()); + Assert.assertEquals(1, lockFiles.size()); + + // 2) check log file content line count == tuples emitted + 1 + List lines = readTextFile(fs, lockFiles.get(0)); + Assert.assertEquals(lines.size(), res.size()+1); + + // 3) read remaining lines in file, then ensure lock is gone + runSpout(spout, "r6"); + lockFiles = listDir(spout.getLockDirPath()); + Assert.assertEquals(0, lockFiles.size()); + + + // 4) --- Create another input file and reverify same behavior --- + Path file2 = new Path(source.toString() + "/file2.txt"); + createTextFile(file2, 10); + + // 5) read initial lines in file, then check if lock exists + res = runSpout(spout, "r5"); + Assert.assertEquals(15, res.size()); + lockFiles = listDir(spout.getLockDirPath()); + Assert.assertEquals(1, lockFiles.size()); + + // 6) check log file content line count == tuples emitted + 1 + lines = readTextFile(fs, lockFiles.get(0)); + Assert.assertEquals(6, lines.size()); + + // 7) read remaining lines in file, then ensure lock is gone + runSpout(spout, "r6"); + lockFiles = listDir(spout.getLockDirPath()); + Assert.assertEquals(0, lockFiles.size()); + } + + @Test + public void testLockLoggingFreqCount() throws Exception { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 10); + + // 0) config spout to log progress in lock file for each tuple + + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(2); // 1 lock log entry every 2 tuples + spout.setCommitFrequencySec(1000); // Effectively disable commits based on time + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + // 1) read 5 lines in file, + runSpout(spout, "r5"); + + // 2) check log file contents + String lockFile = listDir(spout.getLockDirPath()).get(0); + List lines = readTextFile(fs, lockFile); + Assert.assertEquals(lines.size(), 3); + + // 3) read 6th line and see if another log entry was made + runSpout(spout, "r1"); + lines = readTextFile(fs, lockFile); + Assert.assertEquals(lines.size(), 4); + } + + @Test + public void testLockLoggingFreqSec() throws Exception { + Path file1 = new Path(source.toString() + "/file1.txt"); + createTextFile(file1, 10); + + // 0) config spout to log progress in lock file for each tuple + HdfsSpout spout = makeSpout(Configs.TEXT, TextFileReader.defaultFields); + spout.setCommitFrequencyCount(0); // disable it + spout.setCommitFrequencySec(2); // log every 2 sec + + Map conf = getCommonConfigs(); + openSpout(spout, 0, conf); + + // 1) read 5 lines in file + runSpout(spout, "r5"); + + // 2) check log file contents + String lockFile = listDir(spout.getLockDirPath()).get(0); + List lines = readTextFile(fs, lockFile); + Assert.assertEquals(lines.size(), 1); + Thread.sleep(3000); // allow freq_sec to expire + + // 3) read another line and see if another log entry was made + runSpout(spout, "r1"); + lines = readTextFile(fs, lockFile); + Assert.assertEquals(2, lines.size()); + } + + private static List readTextFile(FileSystem fs, String f) throws IOException { + Path file = new Path(f); + FSDataInputStream x = fs.open(file); + BufferedReader reader = new BufferedReader(new InputStreamReader(x)); + String line = null; + ArrayList result = new ArrayList<>(); + while( (line = reader.readLine()) !=null ) + result.add( line ); + return result; + } + + + private Map getCommonConfigs() { + Map conf = new HashMap(); + conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "0"); + return conf; + } + + private HdfsSpout makeSpout(String readerType, String[] outputFields) { + HdfsSpout spout = new HdfsSpout().withOutputFields(outputFields) + .setReaderType(readerType) + .setHdfsUri(hdfsCluster.getURI().toString()) + .setSourceDir(source.toString()) + .setArchiveDir(archive.toString()) + .setBadFilesDir(badfiles.toString()); + + return spout; + } + + private void openSpout(HdfsSpout spout, int spoutId, Map conf) { + MockCollector collector = new MockCollector(); + spout.open(conf, new MockTopologyContext(spoutId), collector); + } + + /** + * Execute a sequence of calls on HdfsSpout. + * + * @param cmds: set of commands to run, + * e.g. "r,r,r,r,a1,f2,...". The commands are: + * r[N] - receive() called N times + * aN - ack, item number: N + * fN - fail, item number: N + */ + + private List runSpout(HdfsSpout spout, String... cmds) { + MockCollector collector = (MockCollector) spout.getCollector(); + for(String cmd : cmds) { + if(cmd.startsWith("r")) { + int count = 1; + if(cmd.length() > 1) { + count = Integer.parseInt(cmd.substring(1)); + } + for(int i=0; i> item = collector.items.get(n); + spout.ack(item.getKey()); + } + else if(cmd.startsWith("f")) { + int n = Integer.parseInt(cmd.substring(1)); + Pair> item = collector.items.get(n); + spout.fail(item.getKey()); + } + } + return collector.lines; + } + + private void createTextFile(Path file, int lineCount) throws IOException { + FSDataOutputStream os = fs.create(file); + int size = 0; + for (int i = 0; i < lineCount; i++) { + os.writeBytes("line " + i + System.lineSeparator()); + String msg = "line " + i + System.lineSeparator(); + size += msg.getBytes().length; + } + os.close(); + } + + + + private static void createSeqFile(FileSystem fs, Path file, int rowCount) throws IOException { + + Configuration conf = new Configuration(); + try { + if(fs.exists(file)) { + fs.delete(file, false); + } + + SequenceFile.Writer w = SequenceFile.createWriter(fs, conf, file, IntWritable.class, Text.class ); + for (int i = 0; i < rowCount; i++) { + w.append(new IntWritable(i), new Text("line " + i)); + } + w.close(); + System.out.println("done"); + } catch (IOException e) { + e.printStackTrace(); + + } + } + + + + static class MockCollector extends SpoutOutputCollector { + //comma separated offsets + public ArrayList lines; + public ArrayList > > items; + + public MockCollector() { + super(null); + lines = new ArrayList<>(); + items = new ArrayList<>(); + } + + + + @Override + public List emit(String streamId, List tuple, Object messageId) { + lines.add(tuple.toString()); + items.add(HdfsUtils.Pair.of(messageId, tuple)); + return null; + } + + @Override + public void emitDirect(int arg0, String arg1, List arg2, Object arg3) { + throw new UnsupportedOperationException("NOT Implemented"); + } + + @Override + public void reportError(Throwable arg0) { + throw new UnsupportedOperationException("NOT Implemented"); + } + + @Override + public long getPendingCount() { + return 0; + } + } // class MockCollector + + + + // Throws IOExceptions for 3rd & 4th call to next(), succeeds on 5th, thereafter + // throws ParseException. Effectively produces 3 lines (1,2 & 3) from each file read + static class MockTextFailingReader extends TextFileReader { + public static final String[] defaultFields = {"line"}; + int readAttempts = 0; + + public MockTextFailingReader(FileSystem fs, Path file, Map conf) throws IOException { + super(fs, file, conf); + } + + @Override + public List next() throws IOException, ParseException { + readAttempts++; + if (readAttempts == 3 || readAttempts ==4) { + throw new IOException("mock test exception"); + } else if (readAttempts > 5 ) { + throw new ParseException("mock test exception", null); + } + return super.next(); + } + } + + static class MockTopologyContext extends TopologyContext { + private final int componentId; + + public MockTopologyContext(int componentId) { + // StormTopology topology, Map stormConf, Map taskToComponent, Map> componentToSortedTasks, Map> componentToStreamToFields, String stormId, String codeDir, String pidDir, Integer taskId, Integer workerPort, List workerTasks, Map defaultResources, Map userResources, Map executorData, Map>> registeredMetrics, Atom openOrPrepareWasCalled + super(null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null); + this.componentId = componentId; + } + + public String getThisComponentId() { + return Integer.toString( componentId ); + } - } + } }