void backoff(int attempt, long deadline) {
        int numRetry = attempt - 1;
        long delay = RETRIES_DELAY_MIN_MS << numRetry;
        if (delay > errorMaxDelayInMillis) {
            delay = ThreadLocalRandom.current().nextLong(errorMaxDelayInMillis);
        }
        long currentTime = time.milliseconds();
        if (delay + currentTime > deadline) {
            delay = Math.max(0, deadline - currentTime);
        }
        log.debug("Sleeping for up to {} millis", delay);
        try {
            stopRequestedLatch.await(delay, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            return;
        }
    }
@Test
    public void testBackoffLimit() throws Exception {
        MockTime time = new MockTime(0, 0, 0);
        CountDownLatch exitLatch = mock(CountDownLatch.class);
        RetryWithToleranceOperator<ConsumerRecord<byte[], byte[]>> retryWithToleranceOperator = new RetryWithToleranceOperator<>(5, 5000, NONE, time, errorHandlingMetrics, exitLatch);

        when(exitLatch.await(300, TimeUnit.MILLISECONDS)).thenAnswer(i -> {
            time.sleep(300);
            return false;
        });
        when(exitLatch.await(600, TimeUnit.MILLISECONDS)).thenAnswer(i -> {
            time.sleep(600);
            return false;
        });
        when(exitLatch.await(1200, TimeUnit.MILLISECONDS)).thenAnswer(i -> {
            time.sleep(1200);
            return false;
        });
        when(exitLatch.await(2400, TimeUnit.MILLISECONDS)).thenAnswer(i -> {
            time.sleep(2400);
            return false;
        });
        when(exitLatch.await(500, TimeUnit.MILLISECONDS)).thenAnswer(i -> {
            time.sleep(500);
            return false;
        });
        when(exitLatch.await(0, TimeUnit.MILLISECONDS)).thenReturn(false);

        retryWithToleranceOperator.backoff(1, 5000);
        retryWithToleranceOperator.backoff(2, 5000);
        retryWithToleranceOperator.backoff(3, 5000);
        retryWithToleranceOperator.backoff(4, 5000);
        retryWithToleranceOperator.backoff(5, 5000);
        // Simulate a small delay between calculating the deadline, and backing off
        time.sleep(1);
        // We may try to begin backing off after the deadline has already passed; make sure
        // that we don't wait with a negative timeout
        retryWithToleranceOperator.backoff(6, 5000);

        verifyNoMoreInteractions(exitLatch);
    }