flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * Triggers a savepoint and two checkpoints. The second checkpoint completes
	 * and subsumes the first checkpoint, but not the first savepoint. Then we
	 * trigger another checkpoint and savepoint. The 2nd savepoint completes and
	 * does neither subsume the last checkpoint nor the first savepoint.
	 */
Triggers a savepoint and two checkpoints. The second checkpoint completes
and subsumes the first checkpoint, but not the first savepoint. Then we
trigger another checkpoint and savepoint. The 2nd savepoint completes and
does neither subsume the last checkpoint nor the first savepoint.
Triggers a savepoint and two checkpoints . The second checkpoint completes and subsumes the first checkpoint but not the first savepoint . Then we trigger another checkpoint and savepoint . The 2nd savepoint completes and does neither subsume the last checkpoint nor the first savepoint .
@Test
	public void testSavepointsAreNotSubsumed() throws Exception {
		final JobID jid = new JobID();
		final long timestamp = System.currentTimeMillis();
		final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
		final ExecutionAttemptID attemptID2 = new ExecutionAttemptID();
		ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
		ExecutionVertex vertex2 = mockExecutionVertex(attemptID2);
		StandaloneCheckpointIDCounter counter = new StandaloneCheckpointIDCounter();
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jid,
			600000,
			600000,
			0,
			Integer.MAX_VALUE,
			ExternalizedCheckpointSettings.none(),
			new ExecutionVertex[] { vertex1, vertex2 },
			new ExecutionVertex[] { vertex1, vertex2 },
			new ExecutionVertex[] { vertex1, vertex2 },
			counter,
			new StandaloneCompletedCheckpointStore(10),
			null,
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		String savepointDir = tmpFolder.newFolder().getAbsolutePath();
		CompletableFuture<CompletedCheckpoint> savepointFuture1 = coord.triggerSavepoint(timestamp, savepointDir);
		long savepointId1 = counter.getLast();
		assertEquals(1, coord.getNumberOfPendingCheckpoints());
		assertTrue(coord.triggerCheckpoint(timestamp + 1, false));
		assertEquals(2, coord.getNumberOfPendingCheckpoints());
		assertTrue(coord.triggerCheckpoint(timestamp + 2, false));
		long checkpointId2 = counter.getLast();
		assertEquals(3, coord.getNumberOfPendingCheckpoints());
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId2));
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId2));
		assertEquals(1, coord.getNumberOfPendingCheckpoints());
		assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
		assertFalse(coord.getPendingCheckpoints().get(savepointId1).isDiscarded());
		assertFalse(savepointFuture1.isDone());
		assertTrue(coord.triggerCheckpoint(timestamp + 3, false));
		long checkpointId3 = counter.getLast();
		assertEquals(2, coord.getNumberOfPendingCheckpoints());
		CompletableFuture<CompletedCheckpoint> savepointFuture2 = coord.triggerSavepoint(timestamp + 4, savepointDir);
		long savepointId2 = counter.getLast();
		assertEquals(3, coord.getNumberOfPendingCheckpoints());
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, savepointId2));
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, savepointId2));
		assertEquals(2, coord.getNumberOfPendingCheckpoints());
		assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
		assertFalse(coord.getPendingCheckpoints().get(savepointId1).isDiscarded());
		assertFalse(savepointFuture1.isDone());
		assertTrue(savepointFuture2.isDone());
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, savepointId1));
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, savepointId1));
		assertEquals(1, coord.getNumberOfPendingCheckpoints());
		assertEquals(1, coord.getNumberOfRetainedSuccessfulCheckpoints());
		assertTrue(savepointFuture1.isDone());
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId3));
		coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId3));
		assertEquals(0, coord.getNumberOfPendingCheckpoints());
		assertEquals(2, coord.getNumberOfRetainedSuccessfulCheckpoints());
	}
@ Test public void testSavepointsAreNotSubsumed ( ) throws Exception { final JobID jid = new JobID ( ) ; final long timestamp = System . currentTimeMillis ( ) ; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID ( ) ; final ExecutionAttemptID attemptID2 = new ExecutionAttemptID ( ) ; ExecutionVertex vertex1 = mockExecutionVertex ( attemptID1 ) ; ExecutionVertex vertex2 = mockExecutionVertex ( attemptID2 ) ; StandaloneCheckpointIDCounter counter = new StandaloneCheckpointIDCounter ( ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( jid , 600000 , 600000 , 0 , Integer . MAX_VALUE , ExternalizedCheckpointSettings . none ( ) , new ExecutionVertex [ ] { vertex1 , vertex2 } , new ExecutionVertex [ ] { vertex1 , vertex2 } , new ExecutionVertex [ ] { vertex1 , vertex2 } , counter , new StandaloneCompletedCheckpointStore ( 10 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; String savepointDir = tmpFolder . newFolder ( ) . getAbsolutePath ( ) ; CompletableFuture < CompletedCheckpoint > savepointFuture1 = coord . triggerSavepoint ( timestamp , savepointDir ) ; long savepointId1 = counter . getLast ( ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp + 1 , false ) ) ; assertEquals ( 2 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp + 2 , false ) ) ; long checkpointId2 = counter . getLast ( ) ; assertEquals ( 3 , coord . getNumberOfPendingCheckpoints ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , checkpointId2 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId2 ) ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 1 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertFalse ( coord . getPendingCheckpoints ( ) . get ( savepointId1 ) . isDiscarded ( ) ) ; assertFalse ( savepointFuture1 . isDone ( ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp + 3 , false ) ) ; long checkpointId3 = counter . getLast ( ) ; assertEquals ( 2 , coord . getNumberOfPendingCheckpoints ( ) ) ; CompletableFuture < CompletedCheckpoint > savepointFuture2 = coord . triggerSavepoint ( timestamp + 4 , savepointDir ) ; long savepointId2 = counter . getLast ( ) ; assertEquals ( 3 , coord . getNumberOfPendingCheckpoints ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , savepointId2 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , savepointId2 ) ) ; assertEquals ( 2 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 1 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertFalse ( coord . getPendingCheckpoints ( ) . get ( savepointId1 ) . isDiscarded ( ) ) ; assertFalse ( savepointFuture1 . isDone ( ) ) ; assertTrue ( savepointFuture2 . isDone ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , savepointId1 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , savepointId1 ) ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 1 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertTrue ( savepointFuture1 . isDone ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , checkpointId3 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId3 ) ) ; assertEquals ( 0 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 2 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; }
Triggers a savepoint and two checkpoints.
Triggers a savepoint and two checkpoints .
// create some mock Execution vertices that receive the checkpoint trigger messages // set up the coordinator and validate the initial state // Trigger savepoint and checkpoint // 2nd checkpoint should subsume the 1st checkpoint, but not the savepoint // 2nd savepoint should not subsume the last checkpoint and the 1st savepoint // Ack first savepoint