@Test public void testStopPeriodicScheduler() throws Exception { final ExecutionAttemptID attemptID1 = new ExecutionAttemptID(); ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( 600000, 600000, 0, Integer.MAX_VALUE, CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION, true, false, 0); CheckpointCoordinator coord = new CheckpointCoordinator( new JobID(), chkConfig, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new StandaloneCheckpointIDCounter(), new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); try { coord.triggerCheckpoint( System.currentTimeMillis(), CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), null, true, false); fail("The triggerCheckpoint call expected an exception"); } catch (CheckpointException e) { assertEquals(CheckpointFailureReason.PERIODIC_SCHEDULER_SHUTDOWN, e.getCheckpointFailureReason()); } try { coord.triggerCheckpoint( System.currentTimeMillis(), CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), null, false, false); } catch (CheckpointException e) { fail("Unexpected exception : " + e.getCheckpointFailureReason().message()); } }
public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { try { triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); return true; } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); return false; } }
CheckpointCoordinator { public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { try { triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); return true; } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); return false; } } }
CheckpointCoordinator { public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { try { triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); return true; } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); return false; } } CheckpointCoordinator(
			JobID job,
			CheckpointCoordinatorConfiguration chkConfig,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			StateBackend checkpointStateBackend,
			Executor executor,
			SharedStateRegistryFactory sharedStateRegistryFactory,
			CheckpointFailureManager failureManager); }
CheckpointCoordinator { public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { try { triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); return true; } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); return false; } } CheckpointCoordinator(
			JobID job,
			CheckpointCoordinatorConfiguration chkConfig,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			StateBackend checkpointStateBackend,
			Executor executor,
			SharedStateRegistryFactory sharedStateRegistryFactory,
			CheckpointFailureManager failureManager); boolean addMasterHook(MasterTriggerRestoreHook<?> hook); int getNumberOfRegisteredMasterHooks(); void setCheckpointStatsTracker(@Nullable CheckpointStatsTracker statsTracker); void shutdown(JobStatus jobStatus); boolean isShutdown(); CompletableFuture<CompletedCheckpoint> triggerSavepoint(
			final long timestamp,
			@Nullable final String targetLocation); CompletableFuture<CompletedCheckpoint> triggerSynchronousSavepoint(
			final long timestamp,
			final boolean advanceToEndOfEventTime,
			@Nullable final String targetLocation); boolean triggerCheckpoint(long timestamp, boolean isPeriodic); @VisibleForTesting PendingCheckpoint triggerCheckpoint(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic,
			boolean advanceToEndOfTime); void receiveDeclineMessage(DeclineCheckpoint message, String taskManagerLocationInfo); boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message, String taskManagerLocationInfo); void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause); boolean restoreLatestCheckpointedState(
			Map<JobVertexID, ExecutionJobVertex> tasks,
			boolean errorIfNoCheckpoint,
			boolean allowNonRestoredState); boolean restoreSavepoint(
			String savepointPointer,
			boolean allowNonRestored,
			Map<JobVertexID, ExecutionJobVertex> tasks,
			ClassLoader userClassLoader); int getNumberOfPendingCheckpoints(); int getNumberOfRetainedSuccessfulCheckpoints(); Map<Long, PendingCheckpoint> getPendingCheckpoints(); List<CompletedCheckpoint> getSuccessfulCheckpoints(); CheckpointStorageCoordinatorView getCheckpointStorage(); CompletedCheckpointStore getCheckpointStore(); CheckpointIDCounter getCheckpointIdCounter(); long getCheckpointTimeout(); boolean isPeriodicCheckpointingConfigured(); void startCheckpointScheduler(); void stopCheckpointScheduler(); void abortPendingCheckpoints(CheckpointException exception); JobStatusListener createActivatorDeactivator(); }
CheckpointCoordinator { public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { try { triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false); return true; } catch (CheckpointException e) { long latestGeneratedCheckpointId = getCheckpointIdCounter().get(); failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId); return false; } } CheckpointCoordinator(
			JobID job,
			CheckpointCoordinatorConfiguration chkConfig,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			StateBackend checkpointStateBackend,
			Executor executor,
			SharedStateRegistryFactory sharedStateRegistryFactory,
			CheckpointFailureManager failureManager); boolean addMasterHook(MasterTriggerRestoreHook<?> hook); int getNumberOfRegisteredMasterHooks(); void setCheckpointStatsTracker(@Nullable CheckpointStatsTracker statsTracker); void shutdown(JobStatus jobStatus); boolean isShutdown(); CompletableFuture<CompletedCheckpoint> triggerSavepoint(
			final long timestamp,
			@Nullable final String targetLocation); CompletableFuture<CompletedCheckpoint> triggerSynchronousSavepoint(
			final long timestamp,
			final boolean advanceToEndOfEventTime,
			@Nullable final String targetLocation); boolean triggerCheckpoint(long timestamp, boolean isPeriodic); @VisibleForTesting PendingCheckpoint triggerCheckpoint(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic,
			boolean advanceToEndOfTime); void receiveDeclineMessage(DeclineCheckpoint message, String taskManagerLocationInfo); boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message, String taskManagerLocationInfo); void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause); boolean restoreLatestCheckpointedState(
			Map<JobVertexID, ExecutionJobVertex> tasks,
			boolean errorIfNoCheckpoint,
			boolean allowNonRestoredState); boolean restoreSavepoint(
			String savepointPointer,
			boolean allowNonRestored,
			Map<JobVertexID, ExecutionJobVertex> tasks,
			ClassLoader userClassLoader); int getNumberOfPendingCheckpoints(); int getNumberOfRetainedSuccessfulCheckpoints(); Map<Long, PendingCheckpoint> getPendingCheckpoints(); List<CompletedCheckpoint> getSuccessfulCheckpoints(); CheckpointStorageCoordinatorView getCheckpointStorage(); CompletedCheckpointStore getCheckpointStore(); CheckpointIDCounter getCheckpointIdCounter(); long getCheckpointTimeout(); boolean isPeriodicCheckpointingConfigured(); void startCheckpointScheduler(); void stopCheckpointScheduler(); void abortPendingCheckpoints(CheckpointException exception); JobStatusListener createActivatorDeactivator();  }