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 a checkpoint and then sends a decline checkpoint message from
	 * one of the tasks. The expected behaviour is that said checkpoint is discarded and a new
	 * checkpoint is triggered.
	 */
This test triggers a checkpoint and then sends a decline checkpoint message from
one of the tasks. The expected behaviour is that said checkpoint is discarded and a new
checkpoint is triggered.
This test triggers a checkpoint and then sends a decline checkpoint message from one of the tasks . The expected behaviour is that said checkpoint is discarded and a new checkpoint is triggered .
@Test
	public void testTriggerAndDeclineCheckpointSimple() {
		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());
			assertTrue(coord.triggerCheckpoint(timestamp, false));
			assertEquals(1, coord.getNumberOfPendingCheckpoints());
			assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
			assertEquals(1, coord.getNumScheduledTasks());
			long checkpointId = coord.getPendingCheckpoints().entrySet().iterator().next().getKey();
			PendingCheckpoint checkpoint = coord.getPendingCheckpoints().get(checkpointId);
			assertNotNull(checkpoint);
			assertEquals(checkpointId, checkpoint.getCheckpointId());
			assertEquals(timestamp, checkpoint.getCheckpointTimestamp());
			assertEquals(jid, checkpoint.getJobId());
			assertEquals(2, checkpoint.getNumberOfNonAcknowledgedTasks());
			assertEquals(0, checkpoint.getNumberOfAcknowledgedTasks());
			assertEquals(0, checkpoint.getOperatorStates().size());
			assertFalse(checkpoint.isDiscarded());
			assertFalse(checkpoint.isFullyAcknowledged());
			verify(vertex1.getCurrentExecutionAttempt()).triggerCheckpoint(checkpointId, timestamp, CheckpointOptions.forCheckpoint());
			verify(vertex2.getCurrentExecutionAttempt()).triggerCheckpoint(checkpointId, timestamp, CheckpointOptions.forCheckpoint());
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId));
			assertEquals(1, checkpoint.getNumberOfAcknowledgedTasks());
			assertEquals(1, checkpoint.getNumberOfNonAcknowledgedTasks());
			assertFalse(checkpoint.isDiscarded());
			assertFalse(checkpoint.isFullyAcknowledged());
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId));
			assertFalse(checkpoint.isDiscarded());
			assertFalse(checkpoint.isFullyAcknowledged());
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID1, checkpointId));
			assertTrue(checkpoint.isDiscarded());
			assertEquals(0, coord.getNumScheduledTasks());
			assertEquals(0, coord.getNumberOfPendingCheckpoints());
			assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID1, checkpointId));
			coord.receiveDeclineMessage(new DeclineCheckpoint(jid, attemptID2, checkpointId));
			assertTrue(checkpoint.isDiscarded());
			coord.shutdown(JobStatus.FINISHED);
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testTriggerAndDeclineCheckpointSimple ( ) { 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 ( ) ) ; assertTrue ( coord . triggerCheckpoint ( timestamp , false ) ) ; assertEquals ( 1 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; assertEquals ( 1 , coord . getNumScheduledTasks ( ) ) ; long checkpointId = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) . next ( ) . getKey ( ) ; PendingCheckpoint checkpoint = coord . getPendingCheckpoints ( ) . get ( checkpointId ) ; assertNotNull ( checkpoint ) ; assertEquals ( checkpointId , checkpoint . getCheckpointId ( ) ) ; assertEquals ( timestamp , checkpoint . getCheckpointTimestamp ( ) ) ; assertEquals ( jid , checkpoint . getJobId ( ) ) ; assertEquals ( 2 , checkpoint . getNumberOfNonAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint . getNumberOfAcknowledgedTasks ( ) ) ; assertEquals ( 0 , checkpoint . getOperatorStates ( ) . size ( ) ) ; assertFalse ( checkpoint . isDiscarded ( ) ) ; assertFalse ( checkpoint . isFullyAcknowledged ( ) ) ; verify ( vertex1 . getCurrentExecutionAttempt ( ) ) . triggerCheckpoint ( checkpointId , timestamp , CheckpointOptions . forCheckpoint ( ) ) ; verify ( vertex2 . getCurrentExecutionAttempt ( ) ) . triggerCheckpoint ( checkpointId , timestamp , CheckpointOptions . forCheckpoint ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId ) ) ; assertEquals ( 1 , checkpoint . getNumberOfAcknowledgedTasks ( ) ) ; assertEquals ( 1 , checkpoint . getNumberOfNonAcknowledgedTasks ( ) ) ; assertFalse ( checkpoint . isDiscarded ( ) ) ; assertFalse ( checkpoint . isFullyAcknowledged ( ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId ) ) ; assertFalse ( checkpoint . isDiscarded ( ) ) ; assertFalse ( checkpoint . isFullyAcknowledged ( ) ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID1 , checkpointId ) ) ; assertTrue ( checkpoint . isDiscarded ( ) ) ; assertEquals ( 0 , coord . getNumScheduledTasks ( ) ) ; assertEquals ( 0 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID1 , checkpointId ) ) ; coord . receiveDeclineMessage ( new DeclineCheckpoint ( jid , attemptID2 , checkpointId ) ) ; assertTrue ( checkpoint . isDiscarded ( ) ) ; coord . shutdown ( JobStatus . FINISHED ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
This test triggers a checkpoint and then sends a decline checkpoint message from
 one of the tasks.
This test triggers a checkpoint and then sends a decline checkpoint message from one of the tasks .
// 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 // validate that we have a pending checkpoint // we have one task scheduled that will cancel after timeout // check that the vertices received the trigger checkpoint message // acknowledge from one of the tasks // acknowledge the same task again (should not matter) // decline checkpoint from the other task, this should cancel the checkpoint // and trigger a new one // the canceler is also removed // validate that we have no new pending checkpoint // decline again, nothing should happen // decline from the other task, nothing should happen