@Override
    public CompletableFuture<Void> getStateConsumedFuture() {
        synchronized (requestLock) {
            List<CompletableFuture<?>> futures = new ArrayList<>(numberOfInputChannels);
            for (InputChannel inputChannel : inputChannels()) {
                if (inputChannel instanceof RecoveredInputChannel) {
                    futures.add(((RecoveredInputChannel) inputChannel).getStateConsumedFuture());
                }
            }
            return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        }
    }
@Test
    void testCheckpointsDeclinedUnlessStateConsumed() {
        SingleInputGate gate = createInputGate(createNettyShuffleEnvironment());
        checkState(!gate.getStateConsumedFuture().isDone());
        assertThatThrownBy(
                        () ->
                                gate.checkpointStarted(
                                        new CheckpointBarrier(
                                                1L,
                                                1L,
                                                alignedNoTimeout(CHECKPOINT, getDefault()))))
                .isInstanceOf(CheckpointException.class);
    }