flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorExternalizedCheckpointsTest.java
/**
	 * Triggers multiple externalized checkpoints and verifies that the metadata
	 * files have been created.
	 */
Triggers multiple externalized checkpoints and verifies that the metadata
files have been created.
Triggers multiple externalized checkpoints and verifies that the metadata files have been created .
@Test
	public void testTriggerAndConfirmSimpleExternalizedCheckpoint()
		throws Exception {
		final JobID jid = new JobID();
		final ExternalizedCheckpointSettings externalizedCheckpointSettings =
			ExternalizedCheckpointSettings.externalizeCheckpoints(false);
		final File checkpointDir = tmp.newFolder();
		final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
		final ExecutionAttemptID attemptID2 = new ExecutionAttemptID();
		ExecutionVertex vertex1 = CheckpointCoordinatorTest.mockExecutionVertex(attemptID1);
		ExecutionVertex vertex2 = CheckpointCoordinatorTest.mockExecutionVertex(attemptID2);
		Map<JobVertexID, ExecutionJobVertex> jobVertices = new HashMap<>();
		jobVertices.put(vertex1.getJobvertexId(), vertex1.getJobVertex());
		jobVertices.put(vertex2.getJobvertexId(), vertex2.getJobVertex());
		CheckpointCoordinator coord = new CheckpointCoordinator(
			jid,
			600000,
			600000,
			0,
			Integer.MAX_VALUE,
			externalizedCheckpointSettings,
			new ExecutionVertex[] { vertex1, vertex2 },
			new ExecutionVertex[] { vertex1, vertex2 },
			new ExecutionVertex[] { vertex1, vertex2 },
			new StandaloneCheckpointIDCounter(),
			new StandaloneCompletedCheckpointStore(1),
			checkpointDir.getAbsolutePath(),
			Executors.directExecutor(),
			SharedStateRegistry.DEFAULT_FACTORY);
		assertEquals(0, coord.getNumberOfPendingCheckpoints());
		assertEquals(0, coord.getNumberOfRetainedSuccessfulCheckpoints());
		{
			final long timestamp1 = System.currentTimeMillis();
			coord.triggerCheckpoint(timestamp1, false);
			long checkpointId1 = coord.getPendingCheckpoints().entrySet().iterator().next()
				.getKey();
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId1));
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId1));
			CompletedCheckpoint latest = coord.getCheckpointStore().getLatestCheckpoint();
			verifyExternalizedCheckpoint(latest, jid, checkpointId1, timestamp1);
			verifyExternalizedCheckpointRestore(latest, jobVertices, vertex1, vertex2);
		}
		{
			final long timestamp2 = System.currentTimeMillis() + 7;
			coord.triggerCheckpoint(timestamp2, false);
			long checkpointId2 = coord.getPendingCheckpoints().entrySet().iterator().next().getKey();
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId2));
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId2));
			CompletedCheckpoint latest = coord.getCheckpointStore().getLatestCheckpoint();
			verifyExternalizedCheckpoint(latest, jid, checkpointId2, timestamp2);
			verifyExternalizedCheckpointRestore(latest, jobVertices, vertex1, vertex2);
		}
		{
			final long timestamp3 = System.currentTimeMillis() + 146;
			coord.triggerCheckpoint(timestamp3, false);
			long checkpointId3 = coord.getPendingCheckpoints().entrySet().iterator().next().getKey();
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID1, checkpointId3));
			coord.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(jid, attemptID2, checkpointId3));
			CompletedCheckpoint latest = coord.getCheckpointStore().getLatestCheckpoint();
			verifyExternalizedCheckpoint(latest, jid, checkpointId3, timestamp3);
			verifyExternalizedCheckpointRestore(latest, jobVertices, vertex1, vertex2);
		}
		coord.shutdown(JobStatus.FINISHED);
	}
@ Test public void testTriggerAndConfirmSimpleExternalizedCheckpoint ( ) throws Exception { final JobID jid = new JobID ( ) ; final ExternalizedCheckpointSettings externalizedCheckpointSettings = ExternalizedCheckpointSettings . externalizeCheckpoints ( false ) ; final File checkpointDir = tmp . newFolder ( ) ; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID ( ) ; final ExecutionAttemptID attemptID2 = new ExecutionAttemptID ( ) ; ExecutionVertex vertex1 = CheckpointCoordinatorTest . mockExecutionVertex ( attemptID1 ) ; ExecutionVertex vertex2 = CheckpointCoordinatorTest . mockExecutionVertex ( attemptID2 ) ; Map < JobVertexID , ExecutionJobVertex > jobVertices = new HashMap < > ( ) ; jobVertices . put ( vertex1 . getJobvertexId ( ) , vertex1 . getJobVertex ( ) ) ; jobVertices . put ( vertex2 . getJobvertexId ( ) , vertex2 . getJobVertex ( ) ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( jid , 600000 , 600000 , 0 , Integer . MAX_VALUE , externalizedCheckpointSettings , new ExecutionVertex [ ] { vertex1 , vertex2 } , new ExecutionVertex [ ] { vertex1 , vertex2 } , new ExecutionVertex [ ] { vertex1 , vertex2 } , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 1 ) , checkpointDir . getAbsolutePath ( ) , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; assertEquals ( 0 , coord . getNumberOfPendingCheckpoints ( ) ) ; assertEquals ( 0 , coord . getNumberOfRetainedSuccessfulCheckpoints ( ) ) ; { final long timestamp1 = System . currentTimeMillis ( ) ; coord . triggerCheckpoint ( timestamp1 , false ) ; long checkpointId1 = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) . next ( ) . getKey ( ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , checkpointId1 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId1 ) ) ; CompletedCheckpoint latest = coord . getCheckpointStore ( ) . getLatestCheckpoint ( ) ; verifyExternalizedCheckpoint ( latest , jid , checkpointId1 , timestamp1 ) ; verifyExternalizedCheckpointRestore ( latest , jobVertices , vertex1 , vertex2 ) ; } { final long timestamp2 = System . currentTimeMillis ( ) + 7 ; coord . triggerCheckpoint ( timestamp2 , false ) ; long checkpointId2 = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) . next ( ) . getKey ( ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , checkpointId2 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId2 ) ) ; CompletedCheckpoint latest = coord . getCheckpointStore ( ) . getLatestCheckpoint ( ) ; verifyExternalizedCheckpoint ( latest , jid , checkpointId2 , timestamp2 ) ; verifyExternalizedCheckpointRestore ( latest , jobVertices , vertex1 , vertex2 ) ; } { final long timestamp3 = System . currentTimeMillis ( ) + 146 ; coord . triggerCheckpoint ( timestamp3 , false ) ; long checkpointId3 = coord . getPendingCheckpoints ( ) . entrySet ( ) . iterator ( ) . next ( ) . getKey ( ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID1 , checkpointId3 ) ) ; coord . receiveAcknowledgeMessage ( new AcknowledgeCheckpoint ( jid , attemptID2 , checkpointId3 ) ) ; CompletedCheckpoint latest = coord . getCheckpointStore ( ) . getLatestCheckpoint ( ) ; verifyExternalizedCheckpoint ( latest , jid , checkpointId3 , timestamp3 ) ; verifyExternalizedCheckpointRestore ( latest , jobVertices , vertex1 , vertex2 ) ; } coord . shutdown ( JobStatus . FINISHED ) ; }
Triggers multiple externalized checkpoints and verifies that the metadata
 files have been created.
Triggers multiple externalized checkpoints and verifies that the metadata files have been created .
// create some mock Execution vertices that receive the checkpoint trigger messages // set up the coordinator and validate the initial state // --------------- // trigger checkpoint 1 // --------------- // --------------- // trigger checkpoint 2 // --------------- // --------------- // trigger checkpoint 3 // ---------------