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 testQueueSizeLimitPriority() {
        final int maxQueuedRequests = 1;
        final boolean isTriggering = true;
        CheckpointRequestDecider decider = decider(maxQueuedRequests);

        CheckpointTriggerRequest checkpoint = regularCheckpoint();
        CheckpointTriggerRequest savepoint = regularSavepoint();

        decider.chooseRequestToExecute(checkpoint, isTriggering, 0);
        decider.chooseRequestToExecute(savepoint, isTriggering, 0);

        assertFailed(checkpoint, TOO_MANY_CHECKPOINT_REQUESTS);
        assertThat(savepoint.getOnCompletionFuture()).isNotDone();
    }