flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * This test verified that after a completed checkpoint a certain time has passed before
	 * another is triggered.
	 */
This test verified that after a completed checkpoint a certain time has passed before
another is triggered.
This test verified that after a completed checkpoint a certain time has passed before another is triggered .
@Test
	public void testMinTimeBetweenCheckpointsInterval() throws Exception {
		final JobID jid = new JobID();
		final ExecutionAttemptID attemptID = new ExecutionAttemptID();
		final ExecutionVertex vertex = mockExecutionVertex(attemptID);
		final Execution executionAttempt = vertex.getCurrentExecutionAttempt();
		final BlockingQueue<Long> triggerCalls = new LinkedBlockingQueue<>();
		doAnswer(invocation -> {
			triggerCalls.add((Long) invocation.getArguments()[0]);
			return null;
		}).when(executionAttempt).triggerCheckpoint(anyLong(), anyLong(), any(CheckpointOptions.class));
		final long delay = 50;
		final CheckpointCoordinator coord = new CheckpointCoordinator(
				jid,
				2,           
				200_000,     
				delay,       
				1,
				ExternalizedCheckpointSettings.none(),
				new ExecutionVertex[] { vertex },
				new ExecutionVertex[] { vertex },
				new ExecutionVertex[] { vertex },
				new StandaloneCheckpointIDCounter(),
				new StandaloneCompletedCheckpointStore(2),
				"dummy-path",
				Executors.directExecutor(),
				SharedStateRegistry.DEFAULT_FACTORY);
		try {
			coord.startCheckpointScheduler();
			Long firstCallId = triggerCalls.take();
			assertEquals(1L, firstCallId.longValue());
			AcknowledgeCheckpoint ackMsg = new AcknowledgeCheckpoint(jid, attemptID, 1L);
			final long ackTime = System.nanoTime();
			coord.receiveAcknowledgeMessage(ackMsg);
			Long nextCallId = triggerCalls.take();
			final long nextCheckpointTime = System.nanoTime();
			assertEquals(2L, nextCallId.longValue());
			final long delayMillis = (nextCheckpointTime - ackTime) / 1_000_000;
			if (delayMillis + 1 < delay) {
				fail("checkpoint came too early: delay was " + delayMillis + " but should have been at least " + delay);
			}
		}
		finally {
			coord.stopCheckpointScheduler();
			coord.shutdown(JobStatus.FINISHED);
		}
	}
@ Test public void testMinTimeBetweenCheckpointsInterval ( ) throws Exception { final JobID jid = new JobID ( ) ; final ExecutionAttemptID attemptID = new ExecutionAttemptID ( ) ; final ExecutionVertex vertex = mockExecutionVertex ( attemptID ) ; final Execution executionAttempt = vertex . getCurrentExecutionAttempt ( ) ; final BlockingQueue < Long > triggerCalls = new LinkedBlockingQueue < > ( ) ; doAnswer ( invocation -> { triggerCalls . add ( ( Long ) invocation . getArguments ( ) [ 0 ] ) ; return null ; } ) . when ( executionAttempt ) . triggerCheckpoint ( anyLong ( ) , anyLong ( ) , any ( CheckpointOptions . class ) ) ; final long delay = 50 ; final CheckpointCoordinator coord = new CheckpointCoordinator ( jid , 2 , 200_000 , delay , 1 , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex } , new ExecutionVertex [ ] { vertex } , new ExecutionVertex [ ] { vertex } , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 2 ) , "dummy-path" , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; try { coord . startCheckpointScheduler ( ) ; Long firstCallId = triggerCalls . take ( ) ; assertEquals ( 1L , firstCallId . longValue ( ) ) ; AcknowledgeCheckpoint ackMsg = new AcknowledgeCheckpoint ( jid , attemptID , 1L ) ; final long ackTime = System . nanoTime ( ) ; coord . receiveAcknowledgeMessage ( ackMsg ) ; Long nextCallId = triggerCalls . take ( ) ; final long nextCheckpointTime = System . nanoTime ( ) ; assertEquals ( 2L , nextCallId . longValue ( ) ) ; final long delayMillis = ( nextCheckpointTime - ackTime ) / 1_000_000 ; if ( delayMillis + 1 < delay ) { fail ( "checkpoint came too early: delay was " + delayMillis + " but should have been at least " + delay ) ; } } finally { coord . stopCheckpointScheduler ( ) ; coord . shutdown ( JobStatus . FINISHED ) ; } }
This test verified that after a completed checkpoint a certain time has passed before
 another is triggered.
This test verified that after a completed checkpoint a certain time has passed before another is triggered .
// create some mock execution vertices and trigger some checkpoint // periodic interval is 2 ms // timeout is very long (200 s) // 50 ms delay between checkpoints // wait until the first checkpoint was triggered // tell the coordinator that the checkpoint is done // wait until the next checkpoint is triggered // we need to add one ms here to account for rounding errors