@Override
    public void write(InputT element, Context context) throws IOException, InterruptedException {
        while (bufferedRequestEntries.size() >= maxBufferedRequests) {
            flush();
        }

        addEntryToBuffer(elementConverter.apply(element, context), false);

        nonBlockingFlush();
    }
@Test
    public void testThatUnwrittenRecordsInBufferArePersistedWhenSnapshotIsTaken()
            throws IOException, InterruptedException {
        AsyncSinkWriterImpl sink =
                new AsyncSinkWriterImplBuilder().context(sinkInitContext).build();
        for (int i = 0; i < 23; i++) {
            sink.write(String.valueOf(i));
        }

        assertThat(res.size()).isEqualTo(20);
        assertThatBufferStatesAreEqual(sink.wrapRequests(20, 21, 22), getWriterState(sink));
    }