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

        return returnRecord;
    }
@Test
    void testReadFirstN() {
        try {
            final String fileContent = "111|222|333|444|555|\n666|777|888|999|000|\n";
            final FileInputSplit split = createTempFile(fileContent);

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

            format.setFieldDelimiter("|");

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

            Tuple2<Integer, Integer> result = new Tuple2<>();

            result = format.nextRecord(result);
            assertThat(result.f0).isEqualTo(Integer.valueOf(111));
            assertThat(result.f1).isEqualTo(Integer.valueOf(222));

            result = format.nextRecord(result);
            assertThat(result.f0).isEqualTo(Integer.valueOf(666));
            assertThat(result.f1).isEqualTo(Integer.valueOf(777));

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