public Task schedule(BackOff backOff, ThrowingFunction<Task, Boolean, Exception> function) {
        final BackOffTimerTask task = new BackOffTimerTask(backOff, scheduler, function);

        long delay = task.next();
        if (delay != BackOff.NEVER) {
            scheduler.schedule(task, delay, TimeUnit.MILLISECONDS);
        } else {
            task.cancel();
        }

        return task;
    }
@Test
    public void testBackOffTimerWithMaxElapsedTime() throws Exception {
        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicInteger counter = new AtomicInteger();
        final ScheduledExecutorService executor = Executors.newScheduledThreadPool(3);
        final BackOff backOff = BackOff.builder().delay(100).maxElapsedTime(400).build();
        final BackOffTimer timer = new BackOffTimer(executor);

        BackOffTimer.Task task = timer.schedule(
                backOff,
                context -> {
                    assertEquals(counter.incrementAndGet(), context.getCurrentAttempts());
                    assertEquals(100, context.getCurrentDelay());
                    assertEquals(100L * counter.get(), context.getCurrentElapsedTime());

                    return true;
                });

        task.whenComplete(
                (context, throwable) -> {
                    assertTrue(counter.get() <= 5);
                    assertEquals(BackOffTimer.Task.Status.Exhausted, context.getStatus());
                    latch.countDown();
                });

        latch.await(5, TimeUnit.SECONDS);
        executor.shutdownNow();
    }