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 testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore() throws Exception {
  		JobID jobId = new JobID();
  		long checkpointTimeout = 60 * 1000;
  		long timestamp = 1272635;
  		ExecutionVertex[] vertices = new ExecutionVertex[] {
  				mockExecutionVertex(jobId),
  				mockExecutionVertex(jobId) };
  		MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
  
  		// Temporary directory for file state backend
  		final File tmpDir = CommonTestUtils.createTempDirectory();
  
  		try {
  			FileSystemStateStore<CompletedCheckpoint> savepointStore = new FileSystemStateStore<>(
  					tmpDir.toURI().toString(), "sp-");
  
  			SavepointCoordinator coordinator = createSavepointCoordinator(
  					jobId,
  					checkpointTimeout,
  					vertices,
  					vertices,
  					vertices,
  					checkpointIdCounter,
  					savepointStore);
  
  			// Trigger the savepoint
  			Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
  			assertFalse(savepointPathFuture.isCompleted());
  
  			long checkpointId = checkpointIdCounter.getLastReturnedCount();
  			assertEquals(0, checkpointId);
  
  			// Verify send trigger messages
  			for (ExecutionVertex vertex : vertices) {
  				verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
  			}
  
  			PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
  					.get(checkpointId);
  
  			verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
  					timestamp, 0, 2, 0, false, false);
  
  			// Acknowledge tasks
  			for (ExecutionVertex vertex : vertices) {
  				coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
  						jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
  						checkpointId, createSerializedStateHandle(vertex), 0));
  			}
  
  			// The pending checkpoint is completed
  			assertTrue(pendingCheckpoint.isDiscarded());
  			assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
  
  			// Verify send notify complete messages
  			for (ExecutionVertex vertex : vertices) {
  				verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
  			}
  
  			// Verify that the future has been completed
  			assertTrue(savepointPathFuture.isCompleted());
  			String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
  
  			// Verify all promises removed
  			assertEquals(0, getSavepointPromises(coordinator).size());
  
  			coordinator.shutdown();
  
  			// Verify the savepoint is still available
  			CompletedCheckpoint savepoint = savepointStore.getState(savepointPath);
  			verifySavepoint(savepoint, jobId, checkpointId, timestamp,
  					vertices);
  		}
  		finally {
  			FileUtils.deleteDirectory(tmpDir);
  		}
  	}
test_case.class_method_signature: SavepointCoordinatorTest.testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore()
test_case.full_signature: @Test public void testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore()
test_case.identifier: testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore
test_case.invocations: mockExecutionVertex mockExecutionVertex createTempDirectory toString toURI createSavepointCoordinator triggerSavepoint assertFalse isCompleted getLastReturnedCount assertEquals verifyTriggerCheckpoint get getPendingCheckpoints verifyPendingCheckpoint receiveAcknowledgeMessage getAttemptId getCurrentExecutionAttempt createSerializedStateHandle assertTrue isDiscarded assertEquals size getSuccessfulCheckpoints verifyNotifyCheckpointComplete assertTrue isCompleted result Zero assertEquals size getSavepointPromises shutdown getState verifySavepoint deleteDirectory
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore()
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)