### Question:
CheckpointFailureManager { public void handleJobLevelCheckpointException(CheckpointException exception, long checkpointId) { checkFailureCounter(exception, checkpointId); if (continuousFailureCounter.get() > tolerableCpFailureNumber) { clearCount(); failureCallback.failJob(new FlinkRuntimeException("Exceeded checkpoint tolerable failure threshold.")); } } CheckpointFailureManager(int tolerableCpFailureNumber, FailJobCallback failureCallback); void handleJobLevelCheckpointException(CheckpointException exception, long checkpointId); void handleTaskLevelCheckpointException(
			CheckpointException exception,
			long checkpointId,
			ExecutionAttemptID executionAttemptID); void checkFailureCounter(
			CheckpointException exception,
			long checkpointId); void handleCheckpointSuccess(long checkpointId); static final int UNLIMITED_TOLERABLE_FAILURE_NUMBER; }### Answer:
@Test public void testContinuousFailure() { TestFailJobCallback callback = new TestFailJobCallback(); CheckpointFailureManager failureManager = new CheckpointFailureManager(2, callback); failureManager.handleJobLevelCheckpointException(new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 1); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 2); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION), 3); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 4); assertEquals(1, callback.getInvokeCounter()); }
@Test public void testTotalCountValue() { TestFailJobCallback callback = new TestFailJobCallback(); CheckpointFailureManager failureManager = new CheckpointFailureManager(0, callback); for (CheckpointFailureReason reason : CheckpointFailureReason.values()) { failureManager.handleJobLevelCheckpointException(new CheckpointException(reason), -1); } assertEquals(1, callback.getInvokeCounter()); }
@Test public void testIgnoreOneCheckpointRepeatedlyCountMultiTimes() { TestFailJobCallback callback = new TestFailJobCallback(); CheckpointFailureManager failureManager = new CheckpointFailureManager(2, callback); failureManager.handleJobLevelCheckpointException(new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 1); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 2); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION), 3); failureManager.handleJobLevelCheckpointException( new CheckpointException(CheckpointFailureReason.CHECKPOINT_DECLINED), 2); assertEquals(0, callback.getInvokeCounter()); }