@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 testThatSnapshotsAreTakenOfBufferCorrectlyBeforeAndAfterAutomaticFlush()
            throws IOException, InterruptedException {
        AsyncSinkWriterImpl sink =
                new AsyncSinkWriterImplBuilder().context(sinkInitContext).maxBatchSize(3).build();

        sink.write("25");
        sink.write("55");

        assertThatBufferStatesAreEqual(sink.wrapRequests(25, 55), getWriterState(sink));
        assertThat(res.size()).isEqualTo(0);

        sink.write("75");

        assertThatBufferStatesAreEqual(BufferedRequestState.emptyState(), getWriterState(sink));
        assertThat(res.size()).isEqualTo(3);
    }