@Override
    public void abortCheckpointOnBarrier(long checkpointId, CheckpointException cause)
            throws IOException {
        if (isCurrentSyncSavepoint(checkpointId)) {
            throw new FlinkRuntimeException("Stop-with-savepoint failed.");
        }
        subtaskCheckpointCoordinator.abortCheckpointOnBarrier(checkpointId, cause, operatorChain);
    }
@Test
    void testSavepointSuspendedAborted() {
        assertThatThrownBy(
                        () ->
                                testSyncSavepointWithEndInput(
                                        (task, id) ->
                                                task.abortCheckpointOnBarrier(
                                                        id,
                                                        new CheckpointException(
                                                                UNKNOWN_TASK_CHECKPOINT_NOTIFICATION_FAILURE)),
                                        SavepointType.suspend(SavepointFormatType.CANONICAL),
                                        false))
                .isInstanceOf(FlinkRuntimeException.class)
                .hasMessage("Stop-with-savepoint failed.");
    }