@Override
    public void fulfillFinishedTaskStatus(Map<OperatorID, OperatorState> operatorStates) {
        if (!mayHaveFinishedTasks) {
            return;
        }

        Map<JobVertexID, ExecutionJobVertex> partlyFinishedVertex = new HashMap<>();
        for (Execution task : finishedTasks) {
            JobVertexID jobVertexId = task.getVertex().getJobvertexId();
            if (!fullyFinishedOrFinishedOnRestoreVertices.containsKey(jobVertexId)) {
                partlyFinishedVertex.put(jobVertexId, task.getVertex().getJobVertex());
            }
        }

        checkNoPartlyFinishedVertexUsedUnionListState(partlyFinishedVertex, operatorStates);
        checkNoPartlyOperatorsFinishedVertexUsedUnionListState(
                partlyFinishedVertex, operatorStates);

        fulfillFullyFinishedOrFinishedOnRestoreOperatorStates(operatorStates);
        fulfillSubtaskStateForPartiallyFinishedOperators(operatorStates);
    }
@Test
    void testFulfillFinishedStates() throws Exception {
        JobVertexID fullyFinishedVertexId = new JobVertexID();
        JobVertexID finishedOnRestoreVertexId = new JobVertexID();
        JobVertexID partiallyFinishedVertexId = new JobVertexID();
        OperatorID fullyFinishedOperatorId = new OperatorID();
        OperatorID finishedOnRestoreOperatorId = new OperatorID();
        OperatorID partiallyFinishedOperatorId = new OperatorID();

        ExecutionGraph executionGraph =
                new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
                        .addJobVertex(
                                fullyFinishedVertexId,
                                2,
                                2,
                                Collections.singletonList(
                                        OperatorIDPair.generatedIDOnly(fullyFinishedOperatorId)),
                                true)
                        .addJobVertex(
                                finishedOnRestoreVertexId,
                                2,
                                2,
                                Collections.singletonList(
                                        OperatorIDPair.generatedIDOnly(
                                                finishedOnRestoreOperatorId)),
                                true)
                        .addJobVertex(
                                partiallyFinishedVertexId,
                                2,
                                2,
                                Collections.singletonList(
                                        OperatorIDPair.generatedIDOnly(
                                                partiallyFinishedOperatorId)),
                                true)
                        .build(EXECUTOR_EXTENSION.getExecutor());
        ExecutionVertex[] fullyFinishedVertexTasks =
                executionGraph.getJobVertex(fullyFinishedVertexId).getTaskVertices();
        ExecutionVertex[] finishedOnRestoreVertexTasks =
                executionGraph.getJobVertex(finishedOnRestoreVertexId).getTaskVertices();
        ExecutionVertex[] partiallyFinishedVertexTasks =
                executionGraph.getJobVertex(partiallyFinishedVertexId).getTaskVertices();
        Arrays.stream(fullyFinishedVertexTasks)
                .forEach(task -> task.getCurrentExecutionAttempt().markFinished());
        partiallyFinishedVertexTasks[0].getCurrentExecutionAttempt().markFinished();

        CheckpointPlan checkpointPlan = createCheckpointPlan(executionGraph);
        Arrays.stream(finishedOnRestoreVertexTasks)
                .forEach(checkpointPlan::reportTaskFinishedOnRestore);

        Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
        checkpointPlan.fulfillFinishedTaskStatus(operatorStates);

        assertThat(operatorStates).hasSize(3);
        assertThat(operatorStates.get(fullyFinishedOperatorId).isFullyFinished()).isTrue();
        assertThat(operatorStates.get(finishedOnRestoreOperatorId).isFullyFinished()).isTrue();
        OperatorState operatorState = operatorStates.get(partiallyFinishedOperatorId);
        assertThat(operatorState.isFullyFinished()).isFalse();
        assertThat(operatorState.getState(0).isFinished()).isTrue();
    }