public SpoutOutputCollector getCollector() {
        return collector;
    }
@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);
        assertEquals(2, listDir(source).size());

        // 2) run spout
        try (
            AutoCloseableHdfsSpout closeableSpout = makeSpout(MockTextFailingReader.class.getName(), MockTextFailingReader.defaultFields)) {
            HdfsSpout spout = closeableSpout.spout;
            Map<String, Object> conf = getCommonConfigs();
            openSpout(spout, 0, conf);

            List<String> res = runSpout(spout, "r11");
            String[] expected = new String[]{ "[line 0]", "[line 1]", "[line 2]", "[line 0]", "[line 1]", "[line 2]" };
            assertArrayEquals(expected, res.toArray());

            // 3) make sure 6 lines (3 from each file) were read in all
            assertEquals(((MockCollector) spout.getCollector()).lines.size(), 6);
            ArrayList<Path> badFiles = HdfsUtils.listFilesByModificationTime(fs, badfiles, 0);
            assertEquals(badFiles.size(), 2);
        }
    }