HdfsSpout extends BaseRichSpout { public SpoutOutputCollector getCollector() { return collector; } 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 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); Map conf = getDefaultConfig(); conf.put(Configs.COMMIT_FREQ_COUNT, "1"); conf.put(Configs.COMMIT_FREQ_SEC, "1"); HdfsSpout spout = makeSpout(0, conf, Configs.TEXT, TextFileReader.defaultFields); 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); Map conf = getDefaultConfig(); conf.put(Configs.COMMIT_FREQ_COUNT, "1"); conf.put(Configs.COMMIT_FREQ_SEC, "1"); conf.put(Config.TOPOLOGY_ACKER_EXECUTORS, "1"); HdfsSpout spout = makeSpout(0, conf, Configs.TEXT, TextFileReader.defaultFields); runSpout(spout, "r6", "a0", "a1", "a2", "a3", "a4"); Path arc1 = new Path(archive.toString() + "/file1.txt"); checkCollectorOutput_txt((MockCollector) spout.getCollector(), arc1); 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 testSimpleSequenceFile() throws IOException { 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); Map conf = getDefaultConfig(); HdfsSpout spout = makeSpout(0, conf, Configs.SEQ, SequenceFileReader.defaultFields); List<String> 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 { 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()); Map conf = getDefaultConfig(); HdfsSpout spout = makeSpout(0, conf, MockTextFailingReader.class.getName(), MockTextFailingReader.defaultFields); List<String> 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()); Assert.assertEquals(((MockCollector) spout.getCollector()).lines.size(), 6); ArrayList<Path> badFiles = HdfsUtils.listFilesByModificationTime(fs, badfiles, 0); Assert.assertEquals(badFiles.size(), 2); }