@Test public void testMinDelayBetweenSavepoints() throws Exception { JobID jobId = new JobID(); final ExecutionAttemptID attemptID1 = new ExecutionAttemptID(); ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); CheckpointCoordinatorConfiguration chkConfig = new CheckpointCoordinatorConfiguration( 100000, 200000, 100000000L, 1, CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION, true, false, 0); CheckpointCoordinator coord = new CheckpointCoordinator( jobId, chkConfig, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new StandaloneCheckpointIDCounter(), new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), Executors.directExecutor(), SharedStateRegistry.DEFAULT_FACTORY, failureManager); String savepointDir = tmpFolder.newFolder().getAbsolutePath(); CompletableFuture<CompletedCheckpoint> savepoint0 = coord.triggerSavepoint(0, savepointDir); assertFalse("Did not trigger savepoint", savepoint0.isDone()); CompletableFuture<CompletedCheckpoint> savepoint1 = coord.triggerSavepoint(1, savepointDir); assertFalse("Did not trigger savepoint", savepoint1.isDone()); }
public CompletableFuture<CompletedCheckpoint> triggerSavepoint( final long timestamp, @Nullable final String targetLocation) { final CheckpointProperties properties = CheckpointProperties.forSavepoint(); return triggerSavepointInternal(timestamp, properties, false, targetLocation); }
CheckpointCoordinator { public CompletableFuture<CompletedCheckpoint> triggerSavepoint( final long timestamp, @Nullable final String targetLocation) { final CheckpointProperties properties = CheckpointProperties.forSavepoint(); return triggerSavepointInternal(timestamp, properties, false, targetLocation); } }
CheckpointCoordinator { public CompletableFuture<CompletedCheckpoint> triggerSavepoint( final long timestamp, @Nullable final String targetLocation) { final CheckpointProperties properties = CheckpointProperties.forSavepoint(); return triggerSavepointInternal(timestamp, properties, false, targetLocation); } 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 CompletableFuture<CompletedCheckpoint> triggerSavepoint( final long timestamp, @Nullable final String targetLocation) { final CheckpointProperties properties = CheckpointProperties.forSavepoint(); return triggerSavepointInternal(timestamp, properties, false, targetLocation); } 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 CompletableFuture<CompletedCheckpoint> triggerSavepoint( final long timestamp, @Nullable final String targetLocation) { final CheckpointProperties properties = CheckpointProperties.forSavepoint(); return triggerSavepointInternal(timestamp, properties, false, targetLocation); } 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();  }