@Override
    public void start(long checkpointId, CheckpointOptions checkpointOptions) {
        LOG.debug("{} starting checkpoint {} ({})", taskName, checkpointId, checkpointOptions);
        ChannelStateWriteResult result = new ChannelStateWriteResult();
        ChannelStateWriteResult put =
                results.computeIfAbsent(
                        checkpointId,
                        id -> {
                            Preconditions.checkState(
                                    results.size() < maxCheckpoints,
                                    String.format(
                                            "%s can't start %d, results.size() > maxCheckpoints: %d > %d",
                                            taskName,
                                            checkpointId,
                                            results.size(),
                                            maxCheckpoints));
                            enqueue(
                                    new CheckpointStartRequest(
                                            jobVertexID,
                                            subtaskIndex,
                                            checkpointId,
                                            result,
                                            checkpointOptions.getTargetLocation()),
                                    false);
                            return result;
                        });
        Preconditions.checkArgument(
                put == result,
                taskName + " result future already present for checkpoint " + checkpointId);
    }
@Test
    void testLimit() throws IOException {
        int maxCheckpoints = 3;
        try (ChannelStateWriterImpl writer =
                new ChannelStateWriterImpl(
                        JOB_VERTEX_ID,
                        TASK_NAME,
                        SUBTASK_INDEX,
                        () -> CHECKPOINT_STORAGE.createCheckpointStorage(JOB_ID),
                        maxCheckpoints,
                        new ChannelStateWriteRequestExecutorFactory(JOB_ID),
                        5)) {
            for (int i = 0; i < maxCheckpoints; i++) {
                writer.start(i, CheckpointOptions.forCheckpointWithDefaultLocation());
            }
            assertThatThrownBy(
                            () ->
                                    writer.start(
                                            maxCheckpoints,
                                            CheckpointOptions.forCheckpointWithDefaultLocation()))
                    .isInstanceOf(IllegalStateException.class);
        }
    }