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 triggers two checkpoints and then sends a decline message from one of the tasks
	 * for the first checkpoint. This should discard the first checkpoint while not triggering
	 * a new checkpoint because a later checkpoint is already in progress.
	 */
This test triggers two checkpoints and then sends a decline message from one of the tasks
for the first checkpoint. This should discard the first checkpoint while not triggering
a new checkpoint because a later checkpoint is already in progress.
This test triggers two checkpoints and then sends a decline message from one of the tasks for the first checkpoint . This should discard the first checkpoint while not triggering a new checkpoint because a later checkpoint is already in progress .
@Test
	public void testTriggerAndDeclineCheckpointComplex() {
		try {
			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);
			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 },
				new StandaloneCheckpointIDCounter(),
				new StandaloneCompletedCheckpointStore(1),
				null,
				Executors.directExecutor(),
				SharedStateRegistry.DEFAULT_FACTORY);
			assertEquals(0, coord.getNumberOfPendingCheckpoints());
			assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
			assertEquals(0, coord.getNumScheduledTasks());
			assertTrue(coord.triggerCheckpoint(timestamp, false));
			assertTrue(coord.triggerCheckpoint(timestamp + 2, false));
			assertEquals(2, coord.getNumberOfPendingCheckpoints());
			assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
			assertEquals(2, coord.getNumScheduledTasks());
			Iterator<Map.Entry<Long, PendingCheckpoint>> it = coord.getPendingCheckpoints().entrySet().iterator();
			long checkpoint1Id = it.next().getKey();
			long checkpoint2Id = it.next().getKey();
			PendingCheckpoint checkpoint1 = coord.getPendingCheckpoints().get(checkpoint1Id);
			PendingCheckpoint checkpoint2 = coord.getPendingCheckpoints().get(checkpoint2Id);
			assertNotNull(checkpoint1);
			assertEquals(checkpoint1Id, checkpoint1.getCheckpointId());
			assertEquals(timestamp, checkpoint1.getCheckpointTimestamp());
			assertEquals(jid, checkpoint1.getJobId());
			assertEquals(2, checkpoint1.getNumberOfNonAcknowledgedTasks());
			assertEquals(0, checkpoint1.getNumberOfAcknowledgedTasks());
			assertEquals(0, checkpoint1.getOperatorStates().size());
			assertFalse(checkpoint1.isDiscarded());
			assertFalse(checkpoint1.isFullyAcknowledged());
			assertNotNull(checkpoint2);
			assertEquals(checkpoint2Id, checkpoint2.getCheckpointId());
			assertEquals(timestamp + 2, checkpoint2.getCheckpointTimestamp());
			assertEquals(jid, checkpoint2.getJobId());
			assertEquals(2, checkpoint2.getNumberOfNonAcknowledgedTasks());
			assertEquals(0, checkpoint2.getNumberOfAcknowledgedTasks());
			assertEquals(0, checkpoint2.getOperatorStates().size());
			assertFalse(checkpoint2.isDiscarded());
			assertFalse(checkpoint2.isFullyAcknowledged());
			{
				verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint1Id), eq(timestamp), any(CheckpointOptions.class));
				verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint1Id), eq(timestamp), any(CheckpointOptions.class));
			}
			 for the second checkpoint
			{
				verify(vertex1.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint2Id), eq(timestamp + 2), any(CheckpointOptions.class));
				verify(vertex2.getCurrentExecutionAttempt(), times(1)).triggerCheckpoint(eq(checkpoint2Id), eq(timestamp + 2), any(CheckpointOptions.class));
			}
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID1, checkpoint1Id));
			assertTrue(checkpoint1.isDiscarded());
			assertEquals(1, coord.getNumberOfPendingCheckpoints());
			assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
			assertEquals(1, coord.getNumScheduledTasks());
			long checkpointIdNew = coord.getPendingCheckpoints().entrySet().iterator().next().getKey();
			PendingCheckpoint checkpointNew = coord.getPendingCheckpoints().get(checkpointIdNew);
			assertEquals(checkpoint2Id, checkpointIdNew);
			assertNotNull(checkpointNew);
			assertEquals(checkpointIdNew, checkpointNew.getCheckpointId());
			assertEquals(jid, checkpointNew.getJobId());
			assertEquals(2, checkpointNew.getNumberOfNonAcknowledgedTasks());
			assertEquals(0, checkpointNew.getNumberOfAcknowledgedTasks());
			assertEquals(0, checkpointNew.getOperatorStates().size());
			assertFalse(checkpointNew.isDiscarded());
			assertFalse(checkpointNew.isFullyAcknowledged());
			assertNotEquals(checkpoint1.getCheckpointId(), checkpointNew.getCheckpointId());
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID1, checkpoint1Id));
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID2, checkpoint1Id));
			assertTrue(checkpoint1.isDiscarded());
			coord.shutdown(JobStatus.FINISHED);
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testTriggerAndDeclineCheckpointComplex ( ) { try { 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 ) ; 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 } , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 1 ) , null , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; assertEquals ( 0 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumScheduledTasks ( ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp , false ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp + 2 , false ) ) ; assertEquals ( 2 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertEquals ( 2 , coord . getNumScheduledTasks ( ) ) ; Iterator < Map . Entry < Long , PendingCheckpoint > > it = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) ; long checkpoint1Id = it . next ( ) . getKey ( ) ; long checkpoint2Id = it . next ( ) . getKey ( ) ; PendingCheckpoint checkpoint1 = coord . getPendingCheckpoints ( ) . get ( checkpoint1Id ) ; PendingCheckpoint checkpoint2 = coord . getPendingCheckpoints ( ) . get ( checkpoint2Id ) ; assertNotNull ( checkpoint1 ) ; assertEquals ( checkpoint1Id , checkpoint1 . getCheckpointId ( ) ) ; assertEquals ( timestamp , checkpoint1 . getCheckpointTimestamp ( ) ) ; assertEquals ( jid , checkpoint1 . getJobId ( ) ) ; assertEquals ( 2 , checkpoint1 . getNumberOfNonAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint1 . getNumberOfAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint1 . getOperatorStates ( ) . size ( ) ) ; assertFalse ( checkpoint1 . isDiscarded ( ) ) ; assertFalse ( checkpoint1 . isFullyAcknowledged ( ) ) ; assertNotNull ( checkpoint2 ) ; assertEquals ( checkpoint2Id , checkpoint2 . getCheckpointId ( ) ) ; assertEquals ( timestamp + 2 , checkpoint2 . getCheckpointTimestamp ( ) ) ; assertEquals ( jid , checkpoint2 . getJobId ( ) ) ; assertEquals ( 2 , checkpoint2 . getNumberOfNonAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint2 . getNumberOfAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint2 . getOperatorStates ( ) . size ( ) ) ; assertFalse ( checkpoint2 . isDiscarded ( ) ) ; assertFalse ( checkpoint2 . isFullyAcknowledged ( ) ) ; { verify ( vertex1 . getCurrentExecutionAttempt ( ) , times ( 1 ) ) . triggerCheckpoint ( eq ( checkpoint1Id ) , eq ( timestamp ) , any ( CheckpointOptions . class ) ) ; verify ( vertex2 . getCurrentExecutionAttempt ( ) , times ( 1 ) ) . triggerCheckpoint ( eq ( checkpoint1Id ) , eq ( timestamp ) , any ( CheckpointOptions . class ) ) ; } { verify ( vertex1 . getCurrentExecutionAttempt ( ) , times ( 1 ) ) . triggerCheckpoint ( eq ( checkpoint2Id ) , eq ( timestamp + 2 ) , any ( CheckpointOptions . class ) ) ; verify ( vertex2 . getCurrentExecutionAttempt ( ) , times ( 1 ) ) . triggerCheckpoint ( eq ( checkpoint2Id ) , eq ( timestamp + 2 ) , any ( CheckpointOptions . class ) ) ; } coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID1 , checkpoint1Id ) ) ; assertTrue ( checkpoint1 . isDiscarded ( ) ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertEquals ( 1 , coord . getNumScheduledTasks ( ) ) ; long checkpointIdNew = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) . next ( ) . getKey ( ) ; PendingCheckpoint checkpointNew = coord . getPendingCheckpoints ( ) . get ( checkpointIdNew ) ; assertEquals ( checkpoint2Id , checkpointIdNew ) ; assertNotNull ( checkpointNew ) ; assertEquals ( checkpointIdNew , checkpointNew . getCheckpointId ( ) ) ; assertEquals ( jid , checkpointNew . getJobId ( ) ) ; assertEquals ( 2 , checkpointNew . getNumberOfNonAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpointNew . getNumberOfAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpointNew . getOperatorStates ( ) . size ( ) ) ; assertFalse ( checkpointNew . isDiscarded ( ) ) ; assertFalse ( checkpointNew . isFullyAcknowledged ( ) ) ; assertNotEquals ( checkpoint1 . getCheckpointId ( ) , checkpointNew . getCheckpointId ( ) ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID1 , checkpoint1Id ) ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID2 , checkpoint1Id ) ) ; assertTrue ( checkpoint1 . isDiscarded ( ) ) ; coord . shutdown ( JobStatus . FINISHED ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
This test triggers two checkpoints and then sends a decline message from one of the tasks
 for the first checkpoint.
This test triggers two checkpoints and then sends a decline message from one of the tasks for the first checkpoint .
// create some mock Execution vertices that receive the checkpoint trigger messages // set up the coordinator and validate the initial state // trigger the first checkpoint. this should succeed // trigger second checkpoint, should also succeed // validate that we have a pending checkpoint // check that the vertices received the trigger checkpoint message // check that the vertices received the trigger checkpoint message for the second checkpoint // decline checkpoint from one of the tasks, this should cancel the checkpoint // validate that we have only one pending checkpoint left // validate that it is the same second checkpoint from earlier // decline again, nothing should happen // decline from the other task, nothing should happen