class FilterRecordInputStream implements Closeable {

    @Nullable
    public FilterRecord<? extends F> nextFilterRecord() throws IOException {
        if (end - position <= HEADER_OVERHEAD) {
            if (end == position) {
                return null;
            }

            throw shortReadFilterHeader(recordFilePath.toString(), (int) (HEADER_OVERHEAD - (end - position)));
        }

        headerBuffer.rewind();
        readFullyOrFail(channel, headerBuffer, position);
        headerBuffer.rewind();

        final int bodyLen = headerBuffer.getInt(BODY_LENGTH_OFFSET);
        if (end - position < HEADER_OVERHEAD + bodyLen) {
            throw shortReadFilterBody(recordFilePath.toString(), (int) ((bodyLen + HEADER_OVERHEAD) - (end - position)));
        }

        checkMagic(headerBuffer);

        final ByteBuffer bodyBuffer = ByteBuffer.allocate(bodyLen);
        readFullyOrFail(channel, bodyBuffer, position + HEADER_OVERHEAD);
        bodyBuffer.flip();

        checkCrc(headerBuffer, bodyBuffer);

        final ByteBuffer filterNameBuffer = readFilterNameBuffer(bodyBuffer);
        final String name = StandardCharsets.UTF_8.decode(filterNameBuffer).toString();
        final F filter = factory.readFrom(new ByteArrayInputStream(bodyBuffer.array(),
                bodyBuffer.position() + filterNameBuffer.limit(), bodyBuffer.remaining()));

        // every thing is fine, we move position forward
        position += bodyLen + HEADER_OVERHEAD;
        return new FilterRecord<>(name, filter);
    }

    public  FilterRecordInputStream(Path recordFilePath, BloomFilterFactory<F, ?> factory);

}