test_class.file: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/SavepointCoordinatorTest.java
test_class.identifier: SavepointCoordinatorTest
test_case.body: @Test
  	public void testAbortSavepointIfAckTasksAreNotExecuted() throws Exception {
  		JobID jobId = new JobID();
  		ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
  				mockExecutionVertex(jobId),
  				mockExecutionVertex(jobId) };
  		ExecutionVertex[] ackVertices = new ExecutionVertex[] {
  				mock(ExecutionVertex.class),
  				mock(ExecutionVertex.class) };
  
  		SavepointCoordinator coordinator = createSavepointCoordinator(
  				jobId,
  				60 * 1000,
  				triggerVertices,
  				ackVertices,
  				new ExecutionVertex[] {},
  				new MockCheckpointIdCounter(),
  				new HeapStateStore<CompletedCheckpoint>());
  
  		// Trigger savepoint
  		Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
  
  		// Abort the savepoint, because the vertices are not running
  		assertTrue(savepointPathFuture.isCompleted());
  
  		try {
  			Await.result(savepointPathFuture, FiniteDuration.Zero());
  			fail("Did not throw expected Exception after shutdown");
  		}
  		catch (Exception ignored) {
  		}
  
  		// Verify all promises removed
  		assertEquals(0, getSavepointPromises(coordinator).size());
  
  		coordinator.shutdown();
  	}
test_case.class_method_signature: SavepointCoordinatorTest.testAbortSavepointIfAckTasksAreNotExecuted()
test_case.full_signature: @Test public void testAbortSavepointIfAckTasksAreNotExecuted()
test_case.identifier: testAbortSavepointIfAckTasksAreNotExecuted
test_case.invocations: mockExecutionVertex mockExecutionVertex mock mock createSavepointCoordinator triggerSavepoint assertTrue isCompleted result Zero fail assertEquals size getSavepointPromises shutdown
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testAbortSavepointIfAckTasksAreNotExecuted()
focal_class.declarator: LOG = LoggerFactory.getLogger(SavepointCoordinator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(SavepointCoordinator.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: savepointStore
focal_class.modifier: private
focal_class.original_string: private StateStore<CompletedCheckpoint> savepointStore;
focal_class.type: StateStore<CompletedCheckpoint>
focal_class.var_name: savepointStore
focal_class.declarator: savepointPromises
focal_class.modifier: private final
focal_class.original_string: private final Map<Long, Promise<String>> savepointPromises;
focal_class.type: Map<Long, Promise<String>>
focal_class.var_name: savepointPromises
focal_class.declarator: savepointRestorePath
focal_class.modifier: private volatile
focal_class.original_string: private volatile String savepointRestorePath;
focal_class.type: String
focal_class.var_name: savepointRestorePath
focal_class.file: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/SavepointCoordinator.java
focal_class.identifier: SavepointCoordinator
focal_class.class_method_signature: SavepointCoordinator.SavepointCoordinator(
  			JobID jobId,
  			long baseInterval,
  			long checkpointTimeout,
  			ExecutionVertex[] tasksToTrigger,
  			ExecutionVertex[] tasksToWaitFor,
  			ExecutionVertex[] tasksToCommitTo,
  			ClassLoader userClassLoader,
  			CheckpointIDCounter checkpointIDCounter,
  			StateStore<CompletedCheckpoint> savepointStore,
  			CheckpointStatsTracker statsTracker)
focal_class.full_signature: public  SavepointCoordinator(
  			JobID jobId,
  			long baseInterval,
  			long checkpointTimeout,
  			ExecutionVertex[] tasksToTrigger,
  			ExecutionVertex[] tasksToWaitFor,
  			ExecutionVertex[] tasksToCommitTo,
  			ClassLoader userClassLoader,
  			CheckpointIDCounter checkpointIDCounter,
  			StateStore<CompletedCheckpoint> savepointStore,
  			CheckpointStatsTracker statsTracker)
focal_class.identifier: SavepointCoordinator
focal_class.modifiers: public
focal_class.parameters: (
  			JobID jobId,
  			long baseInterval,
  			long checkpointTimeout,
  			ExecutionVertex[] tasksToTrigger,
  			ExecutionVertex[] tasksToWaitFor,
  			ExecutionVertex[] tasksToCommitTo,
  			ClassLoader userClassLoader,
  			CheckpointIDCounter checkpointIDCounter,
  			StateStore<CompletedCheckpoint> savepointStore,
  			CheckpointStatsTracker statsTracker)
focal_class.signature:  SavepointCoordinator(
  			JobID jobId,
  			long baseInterval,
  			long checkpointTimeout,
  			ExecutionVertex[] tasksToTrigger,
  			ExecutionVertex[] tasksToWaitFor,
  			ExecutionVertex[] tasksToCommitTo,
  			ClassLoader userClassLoader,
  			CheckpointIDCounter checkpointIDCounter,
  			StateStore<CompletedCheckpoint> savepointStore,
  			CheckpointStatsTracker statsTracker)
focal_class.class_method_signature: SavepointCoordinator.getSavepointRestorePath()
focal_class.full_signature: public String getSavepointRestorePath()
focal_class.identifier: getSavepointRestorePath
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String getSavepointRestorePath()
focal_class.class_method_signature: SavepointCoordinator.triggerSavepoint(long timestamp)
focal_class.full_signature: public Future<String> triggerSavepoint(long timestamp)
focal_class.identifier: triggerSavepoint
focal_class.modifiers: public
focal_class.parameters: (long timestamp)
focal_class.return: Future<String>
focal_class.signature: Future<String> triggerSavepoint(long timestamp)
focal_class.class_method_signature: SavepointCoordinator.restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_class.full_signature: public void restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_class.identifier: restoreSavepoint
focal_class.modifiers: public
focal_class.parameters: (
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_class.return: void
focal_class.signature: void restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_class.class_method_signature: SavepointCoordinator.onShutdown()
focal_class.full_signature: @Override protected void onShutdown()
focal_class.identifier: onShutdown
focal_class.modifiers: @Override protected
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void onShutdown()
focal_class.class_method_signature: SavepointCoordinator.onCancelCheckpoint(long canceledCheckpointId)
focal_class.full_signature: @Override protected void onCancelCheckpoint(long canceledCheckpointId)
focal_class.identifier: onCancelCheckpoint
focal_class.modifiers: @Override protected
focal_class.parameters: (long canceledCheckpointId)
focal_class.return: void
focal_class.signature: void onCancelCheckpoint(long canceledCheckpointId)
focal_class.class_method_signature: SavepointCoordinator.onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.full_signature: @Override protected void onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.identifier: onFullyAcknowledgedCheckpoint
focal_class.modifiers: @Override protected
focal_class.parameters: (CompletedCheckpoint checkpoint)
focal_class.return: void
focal_class.signature: void onFullyAcknowledgedCheckpoint(CompletedCheckpoint checkpoint)
focal_class.class_method_signature: SavepointCoordinator.createActivatorDeactivator(
  			ActorSystem actorSystem,
  			UUID leaderSessionID)
focal_class.full_signature: @Override public ActorGateway createActivatorDeactivator(
  			ActorSystem actorSystem,
  			UUID leaderSessionID)
focal_class.identifier: createActivatorDeactivator
focal_class.modifiers: @Override public
focal_class.parameters: (
  			ActorSystem actorSystem,
  			UUID leaderSessionID)
focal_class.return: ActorGateway
focal_class.signature: ActorGateway createActivatorDeactivator(
  			ActorSystem actorSystem,
  			UUID leaderSessionID)
focal_class.superclass: extends CheckpointCoordinator
focal_method.body: public Future<String> triggerSavepoint(long timestamp) throws Exception {
  		final Promise<String> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();
  
  		try {
  			// Get the checkpoint ID up front. If we fail to trigger the checkpoint,
  			// the ID will have changed, but this is OK as long as the checkpoint ID
  			// generates ascending IDs.
  			final long checkpointId = getAndIncrementCheckpointId();
  
  			if (checkpointId == -1) {
  				throw new IllegalStateException("Failed to get checkpoint Id");
  			}
  
  			// Important: make sure to add the promise to the map before calling
  			// any methods that might trigger callbacks, which require the promise.
  			// Otherwise, the might be race conditions.
  			if (savepointPromises.put(checkpointId, promise) == null) {
  				boolean success = false;
  
  				try {
  					// All good. The future will be completed as soon as the
  					// triggered checkpoint is done.
  					success = triggerCheckpoint(timestamp, checkpointId);
  				}
  				finally {
  					if (!success) {
  						savepointPromises.remove(checkpointId);
  						promise.failure(new Exception("Failed to trigger savepoint"));
  					}
  				}
  			}
  			else {
  				throw new IllegalStateException("Duplicate checkpoint ID");
  			}
  		}
  		catch (Throwable t) {
  			promise.failure(new Exception("Failed to trigger savepoint", t));
  		}
  
  		return promise.future();
  	}
focal_method.class_method_signature: SavepointCoordinator.triggerSavepoint(long timestamp)
focal_method.full_signature: public Future<String> triggerSavepoint(long timestamp)
focal_method.identifier: triggerSavepoint
focal_method.invocations: getAndIncrementCheckpointId put triggerCheckpoint remove failure failure future
focal_method.modifiers: public
focal_method.parameters: (long timestamp)
focal_method.return: Future<String>
focal_method.signature: Future<String> triggerSavepoint(long timestamp)