private void flush() throws InterruptedException {
        RequestInfo requestInfo = createRequestInfo();
        while (rateLimitingStrategy.shouldBlock(requestInfo)) {
            mailboxExecutor.yield();
            requestInfo = createRequestInfo();
        }

        List<RequestEntryT> batch = createNextAvailableBatch(requestInfo);
        if (batch.isEmpty()) {
            return;
        }

        long requestTimestamp = System.currentTimeMillis();

        rateLimitingStrategy.registerInFlightRequest(requestInfo);
        inFlightRequestsCount++;
        submitRequestEntries(
                batch, new AsyncSinkWriterResultHandler(requestTimestamp, batch, requestInfo));
    }
@Test
    public void testRetryableErrorsDoNotViolateAtLeastOnceSemanticsDueToRequeueOfFailures()
            throws IOException, InterruptedException {
        AsyncSinkWriterImpl sink =
                new AsyncSinkWriterImplBuilder()
                        .context(sinkInitContext)
                        .maxBatchSize(3)
                        .maxBatchSizeInBytes(10_000_000)
                        .simulateFailures(true)
                        .build();

        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "25", Arrays.asList(), Arrays.asList(25));

        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "55", Arrays.asList(), Arrays.asList(25, 55));

        // 25, 55 persisted; 965 failed and inflight
        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "965", Arrays.asList(25, 55), Arrays.asList());

        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "75", Arrays.asList(25, 55, 965, 75), Arrays.asList());

        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "95", Arrays.asList(25, 55, 965, 75), Arrays.asList(95));

        /*
         * Writing 955 to the sink increases the buffer to size 3 containing [75, 95, 955]. This
         * triggers the outstanding in flight request with the failed 965 to be run, and 965 is
         * placed at the front of the queue. The failure throttles down {@code maxBatchSize} to 1.
         * buffer now should be [965, 75, 95, 955]
         * A new batch containing 965 is then sent, success causes {@code maxBatchSize} to go up
         * to 3 again.
         * next batch is then created of all requests, 75 and 95 are also persisted.
         * 955 is in flight after failure.
         */
        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "955", Arrays.asList(25, 55, 965, 75), Arrays.asList(95, 955));

        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "550", Arrays.asList(25, 55, 965, 75, 95), Arrays.asList());

        /*
         * [550, 45] are attempted to be persisted
         */
        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "45", Arrays.asList(25, 55, 965, 75, 95, 955, 550), Arrays.asList(45));

        /*
         * [550,45,35] triggers inflight request to be added, buffer should be [955,550,45,35]
         * batch size is reduced to 1.
         * Next request would contain only [995] which is persisted,
         * success causes batch size to rise again to 3. next batch is now [550,45,35].
         * All are persisted and batch size is 3.
         */
        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "35", Arrays.asList(25, 55, 965, 75, 95, 955, 550), Arrays.asList(45, 35));

        /* ] should be in the bufferedRequestEntries
         * [ 550] should be in the inFlightRequest, ready to be added
         * [25, 55, 965, 75, 95, 995, 45, 35] should be downstream already
         */
        writeXToSinkAssertDestinationIsInStateYAndBufferHasZ(
                sink, "535", Arrays.asList(25, 55, 965, 75, 95, 955, 550, 45, 35), Arrays.asList());

        // Checkpoint occurs
        sink.flush(true);

        // Everything is saved
        assertThat(res).isEqualTo(Arrays.asList(25, 55, 965, 75, 95, 955, 550, 45, 35, 535));
        assertThat(getWriterState(sink).getStateSize()).isEqualTo(0);
    }