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 testAbortSavepointIfSubsumed() throws Exception {
  		JobID jobId = new JobID();
  		long checkpointTimeout = 60 * 1000;
  		long[] timestamps = new long[] { 1272635, 1272635 + 10 };
  		long[] checkpointIds = new long[2];
  		ExecutionVertex[] vertices = new ExecutionVertex[] {
  				mockExecutionVertex(jobId),
  				mockExecutionVertex(jobId) };
  		MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
  		HeapStateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
  
  		SavepointCoordinator coordinator = createSavepointCoordinator(
  				jobId,
  				checkpointTimeout,
  				vertices,
  				vertices,
  				vertices,
  				checkpointIdCounter,
  				savepointStore);
  
  		// Trigger the savepoints
  		List<Future<String>> savepointPathFutures = new ArrayList<>();
  
  		savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[0]));
  		checkpointIds[0] = checkpointIdCounter.getLastReturnedCount();
  
  		savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[1]));
  		checkpointIds[1] = checkpointIdCounter.getLastReturnedCount();
  
  		for (Future<String> future : savepointPathFutures) {
  			assertFalse(future.isCompleted());
  		}
  
  		// Verify send trigger messages
  		for (ExecutionVertex vertex : vertices) {
  			verifyTriggerCheckpoint(vertex, checkpointIds[0], timestamps[0]);
  			verifyTriggerCheckpoint(vertex, checkpointIds[1], timestamps[1]);
  		}
  
  		PendingCheckpoint[] pendingCheckpoints = new PendingCheckpoint[] {
  				coordinator.getPendingCheckpoints().get(checkpointIds[0]),
  				coordinator.getPendingCheckpoints().get(checkpointIds[1]) };
  
  		verifyPendingCheckpoint(pendingCheckpoints[0], jobId, checkpointIds[0],
  				timestamps[0], 0, 2, 0, false, false);
  
  		verifyPendingCheckpoint(pendingCheckpoints[1], jobId, checkpointIds[1],
  				timestamps[1], 0, 2, 0, false, false);
  
  		// Acknowledge second checkpoint...
  		for (ExecutionVertex vertex : vertices) {
  			coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
  					jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
  					checkpointIds[1], createSerializedStateHandle(vertex), 0));
  		}
  
  		// ...and one task of first checkpoint
  		coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
  				jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
  				checkpointIds[0], createSerializedStateHandle(vertices[0]), 0));
  
  		// The second pending checkpoint is completed and subsumes the first one
  		assertTrue(pendingCheckpoints[0].isDiscarded());
  		assertTrue(pendingCheckpoints[1].isDiscarded());
  		assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
  
  		// Verify send notify complete messages for second checkpoint
  		for (ExecutionVertex vertex : vertices) {
  			verifyNotifyCheckpointComplete(vertex, checkpointIds[1], timestamps[1]);
  		}
  
  		CompletedCheckpoint[] savepoints = new CompletedCheckpoint[2];
  		String[] savepointPaths = new String[2];
  
  		// Verify that the futures have both been completed
  		assertTrue(savepointPathFutures.get(0).isCompleted());
  
  		try {
  			savepointPaths[0] = Await.result(savepointPathFutures.get(0), FiniteDuration.Zero());
  			fail("Did not throw expected exception");
  		}
  		catch (Exception ignored) {
  		}
  
  		// Verify the second savepoint
  		assertTrue(savepointPathFutures.get(1).isCompleted());
  		savepointPaths[1] = Await.result(savepointPathFutures.get(1), FiniteDuration.Zero());
  		savepoints[1] = savepointStore.getState(savepointPaths[1]);
  		verifySavepoint(savepoints[1], jobId, checkpointIds[1], timestamps[1],
  				vertices);
  
  		// Verify all promises removed
  		assertEquals(0, getSavepointPromises(coordinator).size());
  
  		coordinator.shutdown();
  	}
test_case.class_method_signature: SavepointCoordinatorTest.testAbortSavepointIfSubsumed()
test_case.full_signature: @Test public void testAbortSavepointIfSubsumed()
test_case.identifier: testAbortSavepointIfSubsumed
test_case.invocations: mockExecutionVertex mockExecutionVertex createSavepointCoordinator add triggerSavepoint getLastReturnedCount add triggerSavepoint getLastReturnedCount assertFalse isCompleted verifyTriggerCheckpoint verifyTriggerCheckpoint get getPendingCheckpoints get getPendingCheckpoints verifyPendingCheckpoint verifyPendingCheckpoint receiveAcknowledgeMessage getAttemptId getCurrentExecutionAttempt createSerializedStateHandle receiveAcknowledgeMessage getAttemptId getCurrentExecutionAttempt createSerializedStateHandle assertTrue isDiscarded assertTrue isDiscarded assertEquals size getSuccessfulCheckpoints verifyNotifyCheckpointComplete assertTrue isCompleted get result get Zero fail assertTrue isCompleted get result get Zero getState verifySavepoint assertEquals size getSavepointPromises shutdown
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testAbortSavepointIfSubsumed()
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)