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 savepoints can be triggered concurrently.
	 */
Tests that the savepoints can be triggered concurrently.
Tests that the savepoints can be triggered concurrently .
@Test
	public void testConcurrentSavepoints() throws Exception {
		JobID jobId = new JobID();
		final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
		ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
		StandaloneCheckpointIDCounter checkpointIDCounter = new StandaloneCheckpointIDCounter();
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jobId,
			100000,
			200000,
			0L,
			1, 
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[] { vertex1 },
			new ExecutionVertex[] { vertex1 },
			new ExecutionVertex[] { vertex1 },
			checkpointIDCounter,
			new StandaloneCompletedCheckpointStore(2),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		List<CompletableFuture<CompletedCheckpoint>> savepointFutures = new ArrayList<>();
		int numSavepoints = 5;
		String savepointDir = tmpFolder.newFolder().getAbsolutePath();
		for (int i = 0; i < numSavepoints; i++) {
			savepointFutures.add(coord.triggerSavepoint(i, savepointDir));
		}
		for (CompletableFuture<CompletedCheckpoint> savepointFuture : savepointFutures) {
			assertFalse(savepointFuture.isDone());
		}
		long checkpointId = checkpointIDCounter.getLast();
		for (int i = 0; i < numSavepoints; i++, checkpointId--) {
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jobId, attemptID1, checkpointId));
		}
		for (CompletableFuture<CompletedCheckpoint> savepointFuture : savepointFutures) {
			assertTrue(savepointFuture.isDone());
		}
	}
@ Test public void testConcurrentSavepoints ( ) throws Exception { JobID jobId = new JobID ( ) ; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID ( ) ; ExecutionVertex vertex1 = mockExecutionVertex ( attemptID1 ) ; StandaloneCheckpointIDCounter checkpointIDCounter = new StandaloneCheckpointIDCounter ( ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( jobId , 100000 , 200000 , 0L , 1 , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , checkpointIDCounter , new StandaloneCompletedCheckpointStore ( 2 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; List < CompletableFuture < CompletedCheckpoint > > savepointFutures = new ArrayList < > ( ) ; int numSavepoints = 5 ; String savepointDir = tmpFolder . newFolder ( ) . getAbsolutePath ( ) ; for ( int i = 0 ; i < numSavepoints ; i ++ ) { savepointFutures . add ( coord . triggerSavepoint ( i , savepointDir ) ) ; } for ( CompletableFuture < CompletedCheckpoint > savepointFuture : savepointFutures ) { assertFalse ( savepointFuture . isDone ( ) ) ; } long checkpointId = checkpointIDCounter . getLast ( ) ; for ( int i = 0 ; i < numSavepoints ; i ++ , checkpointId -- ) { coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , attemptID1 , checkpointId ) ) ; } for ( CompletableFuture < CompletedCheckpoint > savepointFuture : savepointFutures ) { assertTrue ( savepointFuture . isDone ( ) ) ; } }
Tests that the savepoints can be triggered concurrently.
Tests that the savepoints can be triggered concurrently .
// max one checkpoint at a time => should not affect savepoints // Trigger savepoints // After triggering multiple savepoints, all should in progress // ACK all savepoints // After ACKs, all should be completed