@Override
    public OUT nextRecord(OUT record) throws IOException {
        OUT returnRecord = null;
        do {
            returnRecord = super.nextRecord(record);
        } while (returnRecord == null && !reachedEnd());

        return returnRecord;
    }
@Test
    void ignoreSingleCharPrefixComments() {
        try {
            final String fileContent =
                    "#description of the data\n"
                            + "#successive commented line\n"
                            + "this is|1|2.0|\n"
                            + "a test|3|4.0|\n"
                            + "#next|5|6.0|\n";

            final FileInputSplit split = createTempFile(fileContent);

            final TupleTypeInfo<Tuple3<String, Integer, Double>> typeInfo =
                    TupleTypeInfo.getBasicTupleTypeInfo(String.class, Integer.class, Double.class);
            final CsvInputFormat<Tuple3<String, Integer, Double>> format =
                    new TupleCsvInputFormat<>(PATH, "\n", "|", typeInfo);
            format.setCommentPrefix("#");

            final Configuration parameters = new Configuration();
            format.configure(parameters);
            format.open(split);

            Tuple3<String, Integer, Double> result = new Tuple3<>();

            result = format.nextRecord(result);
            assertThat(result.f0).isEqualTo("this is");
            assertThat(result.f1).isOne();
            assertThat(result.f2).isEqualTo(new Double(2.0));

            result = format.nextRecord(result);
            assertThat(result.f0).isEqualTo("a test");
            assertThat(result.f1).isEqualTo(Integer.valueOf(3));
            assertThat(result.f2).isEqualTo(new Double(4.0));

            result = format.nextRecord(result);
            assertThat(result).isNull();
        } catch (Exception ex) {
            ex.printStackTrace();
            fail("Test failed due to a " + ex.getClass().getName() + ": " + ex.getMessage());
        }
    }