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 ifTheNumberOfUncompletedInFlightRequestsIsTooManyThenBlockInFlushMethod()
            throws Exception {
        CountDownLatch blockedWriteLatch = new CountDownLatch(1);
        CountDownLatch delayedStartLatch = new CountDownLatch(1);
        AsyncSinkWriterImpl sink =
                new AsyncSinkReleaseAndBlockWriterImpl(
                        sinkInitContextAnyThreadMailbox,
                        1,
                        blockedWriteLatch,
                        delayedStartLatch,
                        false);

        Thread t =
                new Thread(
                        () -> {
                            try {
                                sink.writeAsNonMailboxThread("1");
                                sink.writeAsNonMailboxThread("2");
                                sink.writeAsNonMailboxThread("3");
                            } catch (IOException | InterruptedException e) {
                                e.printStackTrace();
                                fail(
                                        "Auxiliary thread encountered an exception when writing to the sink",
                                        e);
                            }
                        });
        t.start();

        delayedStartLatch.await();
        Thread s =
                new Thread(
                        () -> {
                            try {
                                sink.flush(true);
                                fail(
                                        "Sink did not block successfully and reached here when it shouldn't have.");
                            } catch (InterruptedException ignored) {

                            }
                        });
        Thread.sleep(300);
        assertThat(s.isInterrupted()).isFalse();
        s.interrupt();
        blockedWriteLatch.countDown();

        t.join();

        assertThat(res).isEqualTo(Arrays.asList(1, 2, 3));
    }