flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java
/**
	 * Tests that the externalized checkpoint configuration is respected.
	 */
Tests that the externalized checkpoint configuration is respected.
Tests that the externalized checkpoint configuration is respected .
@Test
	public void testExternalizedCheckpoints() throws Exception {
		try {
			final JobID jid = new JobID();
			final long timestamp = System.currentTimeMillis();
			final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
			ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
			CheckpointCoordinator coord = new CheckpointCoordinator(
				jid,
				600000,
				600000,
				0,
				Integer.MAX_VALUE,
				ExternalizedCheckpointSettings.externalizeCheckpoints(true),
				new ExecutionVertex[] { vertex1 },
				new ExecutionVertex[] { vertex1 },
				new ExecutionVertex[] { vertex1 },
				new StandaloneCheckpointIDCounter(),
				new StandaloneCompletedCheckpointStore(1),
				"fake-directory",
				Executors.directExecutor(),
				SharedStateRegistry.DEFAULT_FACTORY);
			assertTrue(coord.triggerCheckpoint(timestamp, false));
			for (PendingCheckpoint checkpoint : coord.getPendingCheckpoints().values()) {
				CheckpointProperties props = checkpoint.getProps();
				CheckpointProperties expected = CheckpointProperties.forExternalizedCheckpoint(true);
				assertEquals(expected, props);
			}
			coord.shutdown(JobStatus.FINISHED);
		}
		catch (Exception e) {
			e.printStackTrace();
			fail(e.getMessage());
		}
	}
@ Test public void testExternalizedCheckpoints ( ) throws Exception { try { final JobID jid = new JobID ( ) ; final long timestamp = System . currentTimeMillis ( ) ; final ExecutionAttemptID attemptID1 = new ExecutionAttemptID ( ) ; ExecutionVertex vertex1 = mockExecutionVertex ( attemptID1 ) ; CheckpointCoordinator coord = new CheckpointCoordinator ( jid , 600000 , 600000 , 0 , Integer . MAX_VALUE , ExternalizedCheckpointSettings . externalizeCheckpoints ( true ) , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new ExecutionVertex [ ] { vertex1 } , new StandaloneCheckpointIDCounter ( ) , new StandaloneCompletedCheckpointStore ( 1 ) , "fake-directory" , Executors . directExecutor ( ) , SharedStateRegistry . DEFAULT_FACTORY ) ; assertTrue ( coord . triggerCheckpoint ( timestamp , false ) ) ; for ( PendingCheckpoint checkpoint : coord . getPendingCheckpoints ( ) . values ( ) ) { CheckpointProperties props = checkpoint . getProps ( ) ; CheckpointProperties expected = CheckpointProperties . forExternalizedCheckpoint ( true ) ; assertEquals ( expected , props ) ; } coord . shutdown ( JobStatus . FINISHED ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; fail ( e . getMessage ( ) ) ; } }
Tests that the externalized checkpoint configuration is respected.
Tests that the externalized checkpoint configuration is respected .
// create some mock Execution vertices that receive the checkpoint trigger messages // set up the coordinator and validate the initial state // the now we should have a completed checkpoint