@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()); }
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 { 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 { 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); }
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); }
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; }