class RecordPayloadParser implements PayloadParser {
        private final Supplier<RecordParser> recordParserSupplier;
        private final Sinks.Many<Buffer> sink = Reactors.createMany();

        private RecordParser recordParser;

        public RecordPayloadParser(Supplier<RecordParser> recordParserSupplier) {
            this.recordParserSupplier = recordParserSupplier;
            reset();
        }

        @Override
        public void handle(Buffer buffer) {
            recordParser.handle(buffer);
        }

        @Override
        public Flux<Buffer> handlePayload() {
            return sink.asFlux();
        }

        @Override
        public void close() {
            sink.emitComplete(Reactors.emitFailureHandler());
        }

        @Override
        public void reset() {<FILL_FUNCTION_BODY>}
    }

            this.recordParser = recordParserSupplier.get();
            this.recordParser.handler(payload -> {
                sink.emitNext(payload, Reactors.emitFailureHandler());
            });
        
<no_super_class>