@Override
    public OptionalLong apply(OptionalLong previousSendTimeNs) {
        long delayNs;
        if (previousGlobalFailures > 0) {
            // If there were global failures (like a response timeout), we want to wait for the
            // full backoff period.
            delayNs = backoff.backoff(previousGlobalFailures);
        } else if ((numReadyRequests > MAX_ASSIGNMENTS_PER_REQUEST) && !hasInflightRequests) {
            // If there were no previous failures, and we have lots of requests, send it as soon
            // as possible.
            delayNs = 0;
        } else {
            // Otherwise, use the standard delay period. This helps to promote batching, which
            // reduces load on the controller.
            delayNs = backoff.initialInterval();
        }
        long newSendTimeNs = nowNs + delayNs;
        if (previousSendTimeNs.isPresent() && previousSendTimeNs.getAsLong() < newSendTimeNs) {
            // If the previous send time was before the new one we calculated, go with the
            // previous one.
            return previousSendTimeNs;
        }
        // Otherwise, return our new send time.
        return OptionalLong.of(newSendTimeNs);
    }
@Test
    public void applyBackoffIntervalWithExistingEarlierDeadline() {
        assertEquals(OptionalLong.of(BACKOFF.initialInterval() / 2),
            new AssignmentsManagerDeadlineFunction(BACKOFF, 0, 1, false, 12).
                apply(OptionalLong.of(BACKOFF.initialInterval() / 2)));
    }