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

        return returnRecord;
    }
@Test
    void testTailingEmptyFields() throws Exception {
        final String fileContent =
                "aa,bb,cc\n"
                        + // ok
                        "aa,bb,\n"
                        + // the last field is empty
                        "aa,,\n"
                        + // the last two fields are empty
                        ",,\n"
                        + // all fields are empty
                        "aa,bb"; // row too short
        final FileInputSplit split = createTempFile(fileContent);

        final TupleTypeInfo<Tuple3<String, String, String>> typeInfo =
                TupleTypeInfo.getBasicTupleTypeInfo(String.class, String.class, String.class);
        final CsvInputFormat<Tuple3<String, String, String>> format =
                new TupleCsvInputFormat<>(PATH, typeInfo);

        format.setFieldDelimiter(",");

        format.configure(new Configuration());
        format.open(split);

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

        result = format.nextRecord(result);
        assertThat(result.f0).isEqualTo("aa");
        assertThat(result.f1).isEqualTo("bb");
        assertThat(result.f2).isEqualTo("cc");

        result = format.nextRecord(result);
        assertThat(result.f0).isEqualTo("aa");
        assertThat(result.f1).isEqualTo("bb");
        assertThat(result.f2).isEmpty();

        result = format.nextRecord(result);
        assertThat(result.f0).isEqualTo("aa");
        assertThat(result.f1).isEmpty();
        assertThat(result.f2).isEmpty();

        result = format.nextRecord(result);
        assertThat(result.f0).isEmpty();
        assertThat(result.f1).isEmpty();
        assertThat(result.f2).isEmpty();

        try {
            format.nextRecord(result);
            fail("Parse Exception was not thrown! (Row too short)");
        } catch (ParseException e) {
        }
    }