@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 testThatIntermittentlyFailingEntriesShouldBeFlushedWithMainBatchInTimeBasedFlush()
            throws Exception {
        AsyncSinkWriterImpl sink =
                new AsyncSinkWriterImplBuilder()
                        .context(sinkInitContext)
                        .maxBatchSizeInBytes(10_000)
                        .maxTimeInBufferMS(100)
                        .maxRecordSizeInBytes(10_000)
                        .simulateFailures(true)
                        .build();

        TestProcessingTimeService tpts = sinkInitContext.getTestProcessingTimeService();
        tpts.setCurrentTime(0L);
        sink.write("1");
        sink.write("2");
        sink.write("225");
        tpts.setCurrentTime(100L);
        assertThat(res.size()).isEqualTo(2);
        sink.write("3");
        sink.write("4");
        tpts.setCurrentTime(199L);
        assertThat(res.size()).isEqualTo(2);
        tpts.setCurrentTime(200L);
        assertThat(res.size()).isEqualTo(5);
    }