@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 testFulfillFullyFinishedStatesWithCoordinator() throws Exception {
        JobVertexID finishedJobVertexID = new JobVertexID();
        OperatorID finishedOperatorID = new OperatorID();

        ExecutionGraph executionGraph =
                new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
                        .addJobVertex(
                                finishedJobVertexID,
                                1,
                                256,
                                Collections.singletonList(
                                        OperatorIDPair.generatedIDOnly(finishedOperatorID)),
                                true)
                        .build(EXECUTOR_EXTENSION.getExecutor());
        executionGraph
                .getJobVertex(finishedJobVertexID)
                .getTaskVertices()[0]
                .getCurrentExecutionAttempt()
                .markFinished();
        CheckpointPlan checkpointPlan = createCheckpointPlan(executionGraph);

        Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
        OperatorState operatorState = new OperatorState(finishedOperatorID, 1, 256);
        operatorState.setCoordinatorState(new TestingStreamStateHandle());
        operatorStates.put(finishedOperatorID, operatorState);

        checkpointPlan.fulfillFinishedTaskStatus(operatorStates);
        assertThat(operatorStates).hasSize(1);
        assertThat(operatorStates.get(finishedOperatorID).isFullyFinished()).isTrue();
    }