Optional<CheckpointTriggerRequest> chooseRequestToExecute(
            CheckpointTriggerRequest newRequest, boolean isTriggering, long lastCompletionMs) {
        if (queuedRequests.size() >= maxQueuedRequests && !queuedRequests.last().isPeriodic) {
            // there are only non-periodic (ie user-submitted) requests enqueued - retain them and
            // drop the new one
            newRequest.completeExceptionally(new CheckpointException(TOO_MANY_CHECKPOINT_REQUESTS));
            return Optional.empty();
        } else {
            queuedRequests.add(newRequest);
            if (queuedRequests.size() > maxQueuedRequests) {
                queuedRequests
                        .pollLast()
                        .completeExceptionally(
                                new CheckpointException(TOO_MANY_CHECKPOINT_REQUESTS));
            }
            Optional<CheckpointTriggerRequest> request =
                    chooseRequestToExecute(isTriggering, lastCompletionMs);
            request.ifPresent(CheckpointRequestDecider::logInQueueTime);
            return request;
        }
    }
@Test
    void testQueueSizeLimit() {
        final int maxQueuedRequests = 10;
        final boolean isTriggering = true;
        CheckpointRequestDecider decider = decider(maxQueuedRequests);
        List<CheckpointTriggerRequest> requests =
                rangeClosed(0, maxQueuedRequests)
                        .mapToObj(i -> regularCheckpoint())
                        .collect(toList());
        int numAdded = 0;
        for (CheckpointTriggerRequest request : requests) {
            assertThat(decider.chooseRequestToExecute(request, isTriggering, 0)).isNotPresent();
            List<CheckpointTriggerRequest> completed =
                    requests.stream()
                            .filter(r1 -> r1.getOnCompletionFuture().isDone())
                            .collect(toList());
            completed.forEach(r -> assertFailed(r, TOO_MANY_CHECKPOINT_REQUESTS));
            assertThat(completed).hasSize(Math.max(++numAdded - maxQueuedRequests, 0));
        }
    }