void handleJobLevelCheckpointException(
            CheckpointProperties checkpointProperties,
            CheckpointException exception,
            long checkpointId) {
        if (!checkpointProperties.isSavepoint()) {
            checkFailureAgainstCounter(exception, checkpointId, failureCallback::failJob);
        }
    }
@Test
    void testContinuousFailure() {
        TestFailJobCallback callback = new TestFailJobCallback();
        CheckpointFailureManager failureManager = new CheckpointFailureManager(2, callback);
        CheckpointProperties checkpointProperties = forCheckpoint(NEVER_RETAIN_AFTER_TERMINATION);
        failureManager.handleJobLevelCheckpointException(
                checkpointProperties,
                new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED),
                1);
        failureManager.handleJobLevelCheckpointException(
                checkpointProperties,
                new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED),
                2);

        // ignore this
        failureManager.handleJobLevelCheckpointException(
                checkpointProperties,
                new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION),
                3);

        failureManager.handleJobLevelCheckpointException(
                checkpointProperties,
                new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED),
                4);
        assertThat(callback.getInvokeCounter()).isOne();
    }