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 no minimum delay between savepoints is enforced.
	 */
Tests that no minimum delay between savepoints is enforced.
Tests that no minimum delay between savepoints is enforced .
@Test
	public void testMinDelayBetweenSavepoints() throws Exception {
		JobID jobId = new JobID();
		final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
		ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jobId,
			100000,
			200000,
			100000000L, 
			1,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[] { vertex1 },
			new ExecutionVertex[] { vertex1 },
			new ExecutionVertex[] { vertex1 },
			new StandaloneCheckpointIDCounter(),
			new StandaloneCompletedCheckpointStore(2),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		String savepointDir = tmpFolder.newFolder().getAbsolutePath();
		CompletableFuture<CompletedCheckpoint> savepoint0 = coord.triggerSavepoint(0, savepointDir);
		assertFalse("Did not trigger savepoint", savepoint0.isDone());
		CompletableFuture<CompletedCheckpoint> savepoint1 = coord.triggerSavepoint(1, savepointDir);
		assertFalse("Did not trigger savepoint", savepoint1.isDone());
	}
@ Test public void testMinDelayBetweenSavepoints ( ) throws Exception { JobID jobId = new JobID ( ) ; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID ( ) ; ExecutionVertex vertex1 = mockExecutionVertex ( attemptID1 ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( jobId , 100000 , 200000 , 100000000L , 1 , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 2 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; String savepointDir = tmpFolder . newFolder ( ) . getAbsolutePath ( ) ; CompletableFuture < CompletedCheckpoint > savepoint0 = coord . triggerSavepoint ( 0 , savepointDir ) ; assertFalse ( "Did not trigger savepoint" , savepoint0 . isDone ( ) ) ; CompletableFuture < CompletedCheckpoint > savepoint1 = coord . triggerSavepoint ( 1 , savepointDir ) ; assertFalse ( "Did not trigger savepoint" , savepoint1 . isDone ( ) ) ; }
Tests that no minimum delay between savepoints is enforced.
Tests that no minimum delay between savepoints is enforced .
// very long min delay => should not affect savepoints