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 pending checkpoint stats callbacks are created.
	 */
Tests that the pending checkpoint stats callbacks are created.
Tests that the pending checkpoint stats callbacks are created .
@Test
	public void testCheckpointStatsTrackerPendingCheckpointCallback() {
		final long timestamp = System.currentTimeMillis();
		ExecutionVertex vertex1 = mockExecutionVertex(new ExecutionAttemptID());
		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(),
			new StandaloneCompletedCheckpointStore(1),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		CheckpointStatsTracker tracker = mock(CheckpointStatsTracker.class);
		coord.setCheckpointStatsTracker(tracker);
		when(tracker.reportPendingCheckpoint(anyLong(), anyLong(), any(CheckpointProperties.class)))
			.thenReturn(mock(PendingCheckpointStats.class));
		assertTrue(coord.triggerCheckpoint(timestamp, false));
		verify(tracker, times(1))
			.reportPendingCheckpoint(eq(1L), eq(timestamp), eq(CheckpointProperties.forStandardCheckpoint()));
	}
@ Test public void testCheckpointStatsTrackerPendingCheckpointCallback ( ) { final long timestamp = System . currentTimeMillis ( ) ; ExecutionVertex vertex1 = mockExecutionVertex ( new ExecutionAttemptID ( ) ) ; 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 ( ) , new StandaloneCompletedCheckpointStore ( 1 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; CheckpointStatsTracker tracker = mock ( CheckpointStatsTracker . class ) ; coord . setCheckpointStatsTracker ( tracker ) ; when ( tracker . reportPendingCheckpoint ( anyLong ( ) , anyLong ( ) , any ( CheckpointProperties . class ) ) ) . thenReturn ( mock ( PendingCheckpointStats . class ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp , false ) ) ; verify ( tracker , times ( 1 ) ) . reportPendingCheckpoint ( eq ( 1L ) , eq ( timestamp ) , eq ( CheckpointProperties . forStandardCheckpoint ( ) ) ) ; }
Tests that the pending checkpoint stats callbacks are created.
Tests that the pending checkpoint stats callbacks are created .
// set up the coordinator and validate the initial state // Trigger a checkpoint and verify callback