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

        return returnRecord;
    }
@Test
    void testPojoTypeWithTrailingEmptyFields() throws Exception {
        final String fileContent = "123,,3.123,,\n456,BBB,3.23,,";
        final FileInputSplit split = createTempFile(fileContent);

        @SuppressWarnings("unchecked")
        PojoTypeInfo<PrivatePojoItem> typeInfo =
                (PojoTypeInfo<PrivatePojoItem>) TypeExtractor.createTypeInfo(PrivatePojoItem.class);
        CsvInputFormat<PrivatePojoItem> inputFormat = new PojoCsvInputFormat<>(PATH, typeInfo);

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

        PrivatePojoItem item = new PrivatePojoItem();
        inputFormat.nextRecord(item);

        assertThat(item.field1).isEqualTo(123);
        assertThat(item.field2).isEmpty();
        assertThat(item.field3).isEqualTo(Double.valueOf(3.123));
        assertThat(item.field4).isEmpty();

        inputFormat.nextRecord(item);

        assertThat(item.field1).isEqualTo(456);
        assertThat(item.field2).isEqualTo("BBB");
        assertThat(item.field3).isEqualTo(Double.valueOf(3.23));
        assertThat(item.field4).isEmpty();
    }