HdfsSpout extends BaseRichSpout { public Path getLockDirPath() { return lockDirPath; } HdfsSpout(); HdfsSpout withOutputFields(String... fields); HdfsSpout withConfigKey(String configKey); Path getLockDirPath(); SpoutOutputCollector getCollector(); void nextTuple(); void open(Map conf, TopologyContext context, SpoutOutputCollector collector); @Override void ack(Object msgId); @Override void fail(Object msgId); void declareOutputFields(OutputFieldsDeclarer declarer);  }
@Test public void testLocking() throws Exception { Path file1 = new Path(source.toString() + "/file1.txt"); createTextFile(file1, 10); Map conf = getDefaultConfig(); conf.put(Configs.COMMIT_FREQ_COUNT, "1"); conf.put(Configs.COMMIT_FREQ_SEC, "100"); HdfsSpout spout = makeSpout(0, conf, Configs.TEXT, TextFileReader.defaultFields); List<String> res = runSpout(spout, "r5"); Assert.assertEquals(5, res.size()); List<String> lockFiles = listDir(spout.getLockDirPath()); Assert.assertEquals(1, lockFiles.size()); List<String> lines = readTextFile(fs, lockFiles.get(0)); Assert.assertEquals(lines.size(), res.size()+1); runSpout(spout, "r6"); lockFiles = listDir(spout.getLockDirPath()); Assert.assertEquals(0, lockFiles.size()); Path file2 = new Path(source.toString() + "/file2.txt"); createTextFile(file2, 10); res = runSpout(spout, "r5"); Assert.assertEquals(15, res.size()); lockFiles = listDir(spout.getLockDirPath()); Assert.assertEquals(1, lockFiles.size()); lines = readTextFile(fs, lockFiles.get(0)); Assert.assertEquals(6, lines.size()); 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); Map conf = getDefaultConfig(); conf.put(Configs.COMMIT_FREQ_COUNT, "2"); conf.put(Configs.COMMIT_FREQ_SEC, "1000"); HdfsSpout spout = makeSpout(0, conf, Configs.TEXT, TextFileReader.defaultFields); runSpout(spout, "r5"); String lockFile = listDir(spout.getLockDirPath()).get(0); List<String> lines = readTextFile(fs, lockFile); Assert.assertEquals(lines.size(), 3); 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); Map conf = getDefaultConfig(); conf.put(Configs.COMMIT_FREQ_COUNT, "0"); conf.put(Configs.COMMIT_FREQ_SEC, "2"); HdfsSpout spout = makeSpout(0, conf, Configs.TEXT, TextFileReader.defaultFields); runSpout(spout, "r5"); String lockFile = listDir(spout.getLockDirPath()).get(0); List<String> lines = readTextFile(fs, lockFile); Assert.assertEquals(lines.size(), 1); Thread.sleep(3000); runSpout(spout, "r1"); lines = readTextFile(fs, lockFile); Assert.assertEquals(2, lines.size()); }