protected void acknowledgeTask(TaskAcknowledgeOperation ackOperation) {
        final long checkpointId = ackOperation.getBarrier().getId();
        final PendingCheckpoint pendingCheckpoint = pendingCheckpoints.get(checkpointId);
        if (pendingCheckpoint == null) {
            LOG.info("skip already ack checkpoint " + checkpointId);
            return;
        }
        TaskLocation location = ackOperation.getTaskLocation();
        LOG.debug(
                "task[{}]({}/{}) ack. {}",
                location.getTaskID(),
                location.getPipelineId(),
                location.getJobId(),
                ackOperation.getBarrier().toString());

        pendingCheckpoint.acknowledgeTask(
                location,
                ackOperation.getStates(),
                pendingCheckpoint.getCheckpointType().isSavepoint()
                        ? SubtaskStatus.SAVEPOINT_PREPARE_CLOSE
                        : SubtaskStatus.RUNNING);

        if (ackOperation.getBarrier().getCheckpointType().notFinalCheckpoint()
                && ackOperation.getBarrier().prepareClose(location)) {
            completedCloseIdleTask(location);
        }
    }
@Test
    void testACKNotExistPendingCheckpoint() throws CheckpointStorageException {
        CheckpointConfig checkpointConfig = new CheckpointConfig();
        checkpointConfig.setStorage(new CheckpointStorageConfig());
        Map<Integer, CheckpointPlan> planMap = new HashMap<>();
        planMap.put(1, CheckpointPlan.builder().pipelineId(1).build());
        CheckpointManager checkpointManager =
                new CheckpointManager(
                        1L,
                        false,
                        nodeEngine,
                        null,
                        planMap,
                        checkpointConfig,
                        instance.getExecutorService("test"),
                        nodeEngine.getHazelcastInstance().getMap(IMAP_RUNNING_JOB_STATE));
        checkpointManager.acknowledgeTask(
                new TaskAcknowledgeOperation(
                        new TaskLocation(new TaskGroupLocation(1L, 1, 1), 1, 1),
                        new CheckpointBarrier(
                                999, System.currentTimeMillis(), CheckpointType.CHECKPOINT_TYPE),
                        new ArrayList<>()));
    }