flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * FLINK-6328
	 *
	 * Tests that savepoints are not added to the {@link CompletedCheckpointStore} and,
	 * thus, are not subject to job recovery. The reason that we don't want that (until
	 * FLINK-4815 has been finished) is that the lifecycle of savepoints is not controlled
	 * by the {@link CheckpointCoordinator}.
	 */
FLINK-6328
Tests that savepoints are not added to the  CompletedCheckpointStore and,
thus, are not subject to job recovery. The reason that we don't want that (until
FLINK-4815 has been finished) is that the lifecycle of savepoints is not controlled
by the  CheckpointCoordinator.
FLINK - 6328 Tests that savepoints are not added to the CompletedCheckpointStore and thus are not subject to job recovery . The reason that we don ' t want that ( until FLINK - 4815 has been finished ) is that the lifecycle of savepoints is not controlled by the CheckpointCoordinator .
@Test
	public void testSavepointsAreNotAddedToCompletedCheckpointStore() throws Exception {
		final JobID jobId = new JobID();
		final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID();
		final ExecutionVertex vertex1 = mockExecutionVertex(executionAttemptId);
		final CompletedCheckpointStore completedCheckpointStore = new StandaloneCompletedCheckpointStore(1);
		final long checkpointTimestamp1 = 1L;
		final long savepointTimestamp = 2L;
		final long checkpointTimestamp2 = 3L;
		final String savepointDir = tmpFolder.newFolder().getAbsolutePath();
		final StandaloneCheckpointIDCounter checkpointIDCounter = new StandaloneCheckpointIDCounter();
		CheckpointCoordinator checkpointCoordinator = new CheckpointCoordinator(
			jobId,
			600000L,
			600000L,
			0L,
			Integer.MAX_VALUE,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[]{vertex1},
			new ExecutionVertex[]{vertex1},
			new ExecutionVertex[]{vertex1},
			checkpointIDCounter,
			completedCheckpointStore,
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		assertTrue(
			"Triggering of a checkpoint should work.",
			checkpointCoordinator.triggerCheckpoint(checkpointTimestamp1, false));
		assertTrue(0 == completedCheckpointStore.getNumberOfRetainedCheckpoints());
		checkpointCoordinator.receiveAcknowledgeMessage(
			new AcknowledgeCheckpoint(
				jobId,
				executionAttemptId,
				checkpointIDCounter.getLast()));
		assertTrue(1 == completedCheckpointStore.getNumberOfRetainedCheckpoints());
		CompletableFuture<CompletedCheckpoint> savepointFuture = checkpointCoordinator.triggerSavepoint(savepointTimestamp, savepointDir);
		checkpointCoordinator.receiveAcknowledgeMessage(
			new AcknowledgeCheckpoint(
				jobId,
				executionAttemptId,
				checkpointIDCounter.getLast()));
		final CompletedCheckpoint savepoint = savepointFuture.get();
		assertFalse(
			"The savepoint should not have been added to the completed checkpoint store",
			savepoint.getCheckpointID() == completedCheckpointStore.getLatestCheckpoint().getCheckpointID());
		assertTrue(
			"Triggering of a checkpoint should work.",
			checkpointCoordinator.triggerCheckpoint(checkpointTimestamp2, false));
		checkpointCoordinator.receiveAcknowledgeMessage(
			new AcknowledgeCheckpoint(
				jobId,
				executionAttemptId,
				checkpointIDCounter.getLast()));
		assertTrue(
			"The latest completed (proper) checkpoint should have been added to the completed checkpoint store.",
			completedCheckpointStore.getLatestCheckpoint().getCheckpointID() == checkpointIDCounter.getLast());
	}
@ Test public void testSavepointsAreNotAddedToCompletedCheckpointStore ( ) throws Exception { final JobID jobId = new JobID ( ) ; final ExecutionAttemptID executionAttemptId = new ExecutionAttemptID ( ) ; final ExecutionVertex vertex1 = mockExecutionVertex ( executionAttemptId ) ; final CompletedCheckpointStore completedCheckpointStore = new StandaloneCompletedCheckpointStore ( 1 ) ; final long checkpointTimestamp1 = 1L ; final long savepointTimestamp = 2L ; final long checkpointTimestamp2 = 3L ; final String savepointDir = tmpFolder . newFolder ( ) . getAbsolutePath ( ) ; final StandaloneCheckpointIDCounter checkpointIDCounter = new StandaloneCheckpointIDCounter ( ) ; CheckpointCoordinator checkpointCoordinator = new CheckpointCoordinator ( jobId , 600000L , 600000L , 0L , Integer . MAX_VALUE , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , checkpointIDCounter , completedCheckpointStore , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; assertTrue ( "Triggering of a checkpoint should work." , checkpointCoordinator . triggerCheckpoint ( checkpointTimestamp1 , false ) ) ; assertTrue ( 0 == completedCheckpointStore . getNumberOfRetainedCheckpoints ( ) ) ; checkpointCoordinator . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , executionAttemptId , checkpointIDCounter . getLast ( ) ) ) ; assertTrue ( 1 == completedCheckpointStore . getNumberOfRetainedCheckpoints ( ) ) ; CompletableFuture < CompletedCheckpoint > savepointFuture = checkpointCoordinator . triggerSavepoint ( savepointTimestamp , savepointDir ) ; checkpointCoordinator . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , executionAttemptId , checkpointIDCounter . getLast ( ) ) ) ; final CompletedCheckpoint savepoint = savepointFuture . get ( ) ; assertFalse ( "The savepoint should not have been added to the completed checkpoint store" , savepoint . getCheckpointID ( ) == completedCheckpointStore . getLatestCheckpoint ( ) . getCheckpointID ( ) ) ; assertTrue ( "Triggering of a checkpoint should work." , checkpointCoordinator . triggerCheckpoint ( checkpointTimestamp2 , false ) ) ; checkpointCoordinator . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jobId , executionAttemptId , checkpointIDCounter . getLast ( ) ) ) ; assertTrue ( "The latest completed (proper) checkpoint should have been added to the completed checkpoint store." , completedCheckpointStore . getLatestCheckpoint ( ) . getCheckpointID ( ) == checkpointIDCounter . getLast ( ) ) ; }
FLINK-6328
 Tests that savepoints are not added to the {@link CompletedCheckpointStore} and,
 thus, are not subject to job recovery.
FLINK - 6328 Tests that savepoints are not added to the { @link CompletedCheckpointStore } and thus are not subject to job recovery .
// trigger a first checkpoint // complete the 1st checkpoint // check that the checkpoint has been completed // trigger a savepoint --> this should not have any effect on the CompletedCheckpointStore // check that no errors occurred // complete the 2nd checkpoint