flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * Tests that the restore callbacks are called if registered.
	 */
Tests that the restore callbacks are called if registered.
Tests that the restore callbacks are called if registered .
@Test
	public void testCheckpointStatsTrackerRestoreCallback() throws Exception {
		ExecutionVertex vertex1 = mockExecutionVertex(new ExecutionAttemptID());
		StandaloneCompletedCheckpointStore store = new StandaloneCompletedCheckpointStore(1);
		CheckpointCoordinator coord = new CheckpointCoordinator(
			new JobID(),
			600000,
			600000,
			0,
			Integer.MAX_VALUE,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[]{vertex1},
			new ExecutionVertex[]{vertex1},
			new ExecutionVertex[]{vertex1},
			new StandaloneCheckpointIDCounter(),
			store,
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		store.addCheckpoint(new CompletedCheckpoint(
			new JobID(),
			0,
			0,
			0,
			Collections.<OperatorID, OperatorState>emptyMap(),
			Collections.<MasterState>emptyList(),
			CheckpointProperties.forStandardCheckpoint(),
			null,
			null));
		CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class);
		coord.setCheckpointStatsTracker(tracker);
		assertTrue(coord.restoreLatestCheckpointedState(Collections.<JobVertexID, ExecutionJobVertex>emptyMap(), false, true));
		verify(tracker, times(1))
			.reportRestoredCheckpoint(any(RestoredCheckpointStats.class));
	}
@ Test public void testCheckpointStatsTrackerRestoreCallback ( ) throws Exception { ExecutionVertex vertex1 = mockExecutionVertex ( new ExecutionAttemptID ( ) ) ; StandaloneCompletedCheckpointStore store = new StandaloneCompletedCheckpointStore ( 1 ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( new JobID ( ) , 600000 , 600000 , 0 , Integer . MAX_VALUE , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new StandaloneCheckpointIDCounter ( ) , store , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; store . addCheckpoint ( new CompletedCheckpoint ( new JobID ( ) , 0 , 0 , 0 , Collections . < OperatorID , OperatorState > emptyMap ( ) , Collections . < MasterState > emptyList ( ) , CheckpointProperties . forStandardCheckpoint ( ) , null , null ) ) ; CheckpointStatsTracker tracker = mock ( CheckpointStatsTracker . class ) ; coord . setCheckpointStatsTracker ( tracker ) ; assertTrue ( coord . restoreLatestCheckpointedState ( Collections . < JobVertexID , ExecutionJobVertex > emptyMap ( ) , false , true ) ) ; verify ( tracker , times ( 1 ) ) . reportRestoredCheckpoint ( any ( RestoredCheckpointStats . class ) ) ; }
Tests that the restore callbacks are called if registered.
Tests that the restore callbacks are called if registered .
// set up the coordinator and validate the initial state