@Override
    public void handleExecutionsTermination(Collection<ExecutionState> terminatedExecutionStates) {
        final Set<ExecutionState> notFinishedExecutionStates =
                checkNotNull(terminatedExecutionStates).stream()
                        .filter(state -> state != ExecutionState.FINISHED)
                        .collect(Collectors.toSet());

        if (notFinishedExecutionStates.isEmpty()) {
            handleExecutionsFinished();
        } else {
            handleAnyExecutionNotFinished(notFinishedExecutionStates);
        }
    }
@Test
    void testSavepointCreationFailureWithFailingExecutions() {
        // no global fail-over is expected to be triggered by the stop-with-savepoint despite the
        // execution failure
        assertSavepointCreationFailure(
                testInstance ->
                        testInstance.handleExecutionsTermination(
                                Collections.singletonList(ExecutionState.FAILED)));
    }