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

        return returnRecord;
    }
@Test
    void testEmptyFields() {
        try {
            final String fileContent =
                    "|0|0|0|0|0|\n"
                            + "1||1|1|1|1|\n"
                            + "2|2||2|2|2|\n"
                            + "3|3|3| |3|3|\n"
                            + "4|4|4|4||4|\n"
                            + "5|5|5|5|5||\n";
            final FileInputSplit split = createTempFile(fileContent);

            final TupleTypeInfo<Tuple6<Short, Integer, Long, Float, Double, Byte>> typeInfo =
                    TupleTypeInfo.getBasicTupleTypeInfo(
                            Short.class,
                            Integer.class,
                            Long.class,
                            Float.class,
                            Double.class,
                            Byte.class);
            final CsvInputFormat<Tuple6<Short, Integer, Long, Float, Double, Byte>> format =
                    new TupleCsvInputFormat<>(PATH, typeInfo);

            format.setFieldDelimiter("|");

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

            Tuple6<Short, Integer, Long, Float, Double, Byte> result = new Tuple6<>();

            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (ShortParser)");
            } catch (ParseException e) {
            }
            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (IntegerParser)");
            } catch (ParseException e) {
            }
            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (LongParser)");
            } catch (ParseException e) {
            }
            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (FloatParser)");
            } catch (ParseException e) {
            }
            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (DoubleParser)");
            } catch (ParseException e) {
            }
            try {
                result = format.nextRecord(result);
                fail("Empty String Parse Exception was not thrown! (ByteParser)");
            } catch (ParseException e) {
            }

            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());
        }
    }