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 testRollbackSetsCheckpointID() throws Exception {
  		CompletedCheckpoint savepoint = mock(CompletedCheckpoint.class);
  		when(savepoint.getStates()).thenReturn(Collections.<StateForTask>emptyList());
  		when(savepoint.getCheckpointID()).thenReturn(12312312L);
  
  		CheckpointIDCounter checkpointIdCounter = mock(CheckpointIDCounter.class);
  
  		StateStore<CompletedCheckpoint> savepointStore = mock(StateStore.class);
  		when(savepointStore.getState(anyString())).thenReturn(savepoint);
  
  		SavepointCoordinator coordinator = createSavepointCoordinator(
  				new JobID(),
  				60 * 1000,
  				new ExecutionVertex[] {},
  				new ExecutionVertex[] {},
  				new ExecutionVertex[] {},
  				checkpointIdCounter,
  				savepointStore);
  
  		coordinator.restoreSavepoint(createExecutionJobVertexMap(), "any");
  
  		verify(checkpointIdCounter).setCount(eq(12312312L + 1));
  
  		coordinator.shutdown();
  	}
test_case.class_method_signature: SavepointCoordinatorTest.testRollbackSetsCheckpointID()
test_case.full_signature: @Test public void testRollbackSetsCheckpointID()
test_case.identifier: testRollbackSetsCheckpointID
test_case.invocations: mock thenReturn when getStates emptyList thenReturn when getCheckpointID mock mock thenReturn when getState anyString createSavepointCoordinator restoreSavepoint createExecutionJobVertexMap setCount verify eq shutdown
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRollbackSetsCheckpointID()
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 void restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath) throws Exception {
  
  		checkNotNull(savepointPath, "Savepoint path");
  
  		synchronized (lock) {
  			if (isShutdown()) {
  				throw new IllegalStateException("CheckpointCoordinator is shut down");
  			}
  
  			long recoveryTimestamp = System.currentTimeMillis();
  
  			LOG.info("Rolling back to savepoint '{}'.", savepointPath);
  
  			CompletedCheckpoint checkpoint = savepointStore.getState(savepointPath);
  
  			LOG.info("Savepoint: {}@{}", checkpoint.getCheckpointID(), checkpoint.getTimestamp());
  
  			// Set the initial state of all tasks
  			LOG.debug("Rolling back individual operators.");
  			for (StateForTask state : checkpoint.getStates()) {
  				LOG.debug("Rolling back subtask {} of operator {}.",
  						state.getSubtask(), state.getOperatorId());
  
  				ExecutionJobVertex vertex = tasks.get(state.getOperatorId());
  
  				if (vertex == null) {
  					String msg = String.format("Failed to rollback to savepoint %s. " +
  							"Cannot map old state for task %s to the new program. " +
  							"This indicates that the program has been changed in a " +
  							"non-compatible way  after the savepoint.", checkpoint,
  							state.getOperatorId());
  					throw new IllegalStateException(msg);
  				}
  
  				if (state.getSubtask() >= vertex.getParallelism()) {
  					String msg = String.format("Failed to rollback to savepoint %s. " +
  							"Parallelism mismatch between savepoint state and new program. " +
  							"Cannot map subtask %d of operator %s to new program with " +
  							"parallelism %d. This indicates that the program has been changed " +
  							"in a non-compatible way after the savepoint.", checkpoint,
  							state.getSubtask(), state.getOperatorId(), vertex.getParallelism());
  					throw new IllegalStateException(msg);
  				}
  
  				Execution exec = vertex.getTaskVertices()[state.getSubtask()]
  						.getCurrentExecutionAttempt();
  
  				exec.setInitialState(state.getState(), recoveryTimestamp);
  			}
  
  			// Reset the checkpoint ID counter
  			long nextCheckpointId = checkpoint.getCheckpointID();
  			checkpointIdCounter.start();
  			checkpointIdCounter.setCount(nextCheckpointId + 1);
  			LOG.info("Reset the checkpoint ID to {}", nextCheckpointId);
  
  			if (savepointRestorePath == null) {
  				savepointRestorePath = savepointPath;
  			}
  		}
  	}
focal_method.class_method_signature: SavepointCoordinator.restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_method.full_signature: public void restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_method.identifier: restoreSavepoint
focal_method.invocations: checkNotNull isShutdown currentTimeMillis info getState info getCheckpointID getTimestamp debug getStates debug getSubtask getOperatorId get getOperatorId format getOperatorId getSubtask getParallelism format getSubtask getOperatorId getParallelism getCurrentExecutionAttempt getTaskVertices getSubtask setInitialState getState getCheckpointID start setCount info
focal_method.modifiers: public
focal_method.parameters: (
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)
focal_method.return: void
focal_method.signature: void restoreSavepoint(
  			Map<JobVertexID, ExecutionJobVertex> tasks,
  			String savepointPath)