@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 testAbortionIfPartiallyFinishedVertexUsedUnionListState() throws Exception {
        JobVertexID jobVertexId = new JobVertexID();
        OperatorID operatorId = new OperatorID();

        ExecutionGraph executionGraph =
                new CheckpointCoordinatorTestingUtils.CheckpointExecutionGraphBuilder()
                        .addJobVertex(
                                jobVertexId,
                                2,
                                2,
                                Collections.singletonList(
                                        OperatorIDPair.generatedIDOnly(operatorId)),
                                true)
                        .build(EXECUTOR_EXTENSION.getExecutor());
        ExecutionVertex[] tasks = executionGraph.getJobVertex(jobVertexId).getTaskVertices();
        tasks[0].getCurrentExecutionAttempt().markFinished();

        CheckpointPlan checkpointPlan = createCheckpointPlan(executionGraph);

        Map<OperatorID, OperatorState> operatorStates = new HashMap<>();
        OperatorState operatorState = new OperatorState(operatorId, 2, 2);
        operatorState.putState(
                0, createSubtaskStateWithUnionListState(TempDirUtils.newFile(temporaryFolder)));
        operatorStates.put(operatorId, operatorState);

        assertThatThrownBy(() -> checkpointPlan.fulfillFinishedTaskStatus(operatorStates))
                .hasMessage(
                        String.format(
                                "The vertex %s (id = %s) has "
                                        + "used UnionListState, but part of its tasks are FINISHED.",
                                executionGraph.getJobVertex(jobVertexId).getName(), jobVertexId))
                .isInstanceOf(FlinkRuntimeException.class);
    }