flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/savepoint/SavepointStoreTest.java
/**
	 * Tests addition of a new savepoint version.
	 */
Tests addition of a new savepoint version.
Tests addition of a new savepoint version .
@Test
	@SuppressWarnings("unchecked")
	public void testMultipleSavepointVersions() throws Exception {
		Field field = SavepointSerializers.class.getDeclaredField("SERIALIZERS");
		field.setAccessible(true);
		Map<Integer, SavepointSerializer<?>> serializers = (Map<Integer, SavepointSerializer<?>>) field.get(null);
		assertTrue(serializers.size() >= 1);
		String root = tmp.getRoot().getAbsolutePath();
		File rootFile = new File(root);
		int version = ThreadLocalRandom.current().nextInt(Integer.MAX_VALUE); 
		long checkpointId = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE); 
		serializers.put(version, NewSavepointSerializer.INSTANCE);
		String savepointDirectory1 = SavepointStore.createSavepointDirectory(root, new JobID());
		TestSavepoint newSavepoint = new TestSavepoint(version, checkpointId);
		String pathNewSavepoint = SavepointStore.storeSavepoint(savepointDirectory1, newSavepoint);
		File[] list = rootFile.listFiles();
		assertNotNull(list);
		assertEquals(1, list.length);
		String savepointDirectory2 = SavepointStore.createSavepointDirectory(root, new JobID());
		SavepointV2 savepoint = new SavepointV2(
			checkpointId,
			CheckpointTestUtils.createOperatorStates(4, 32),
			Collections.<MasterState>emptyList());
		String pathSavepoint = SavepointStore.storeSavepoint(savepointDirectory2, savepoint);
		list = rootFile.listFiles();
		assertNotNull(list);
		assertEquals(2, list.length);
		Savepoint loaded = SavepointStore.loadSavepoint(pathNewSavepoint, Thread.currentThread().getContextClassLoader());
		assertEquals(newSavepoint, loaded);
		loaded = SavepointStore.loadSavepoint(pathSavepoint, Thread.currentThread().getContextClassLoader());
		assertEquals(savepoint.getCheckpointId(), loaded.getCheckpointId());
		assertEquals(savepoint.getTaskStates(), loaded.getTaskStates());
		assertEquals(savepoint.getMasterStates(), loaded.getMasterStates());
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testMultipleSavepointVersions ( ) throws Exception { Field field = SavepointSerializers . class . getDeclaredField ( "SERIALIZERS" ) ; field . setAccessible ( true ) ; Map < Integer , SavepointSerializer < ? > > serializers = ( Map < Integer , SavepointSerializer < ? > > ) field . get ( null ) ; assertTrue ( serializers . size ( ) >= 1 ) ; String root = tmp . getRoot ( ) . getAbsolutePath ( ) ; File rootFile = new File ( root ) ; int version = ThreadLocalRandom . current ( ) . nextInt ( Integer . MAX_VALUE ) ; long checkpointId = ThreadLocalRandom . current ( ) . nextLong ( Long . MAX_VALUE ) ; serializers . put ( version , NewSavepointSerializer . INSTANCE ) ; String savepointDirectory1 = SavepointStore . createSavepointDirectory ( root , new JobID ( ) ) ; TestSavepoint newSavepoint = new TestSavepoint ( version , checkpointId ) ; String pathNewSavepoint = SavepointStore . storeSavepoint ( savepointDirectory1 , newSavepoint ) ; File [ ] list = rootFile . listFiles ( ) ; assertNotNull ( list ) ; assertEquals ( 1 , list . length ) ; String savepointDirectory2 = SavepointStore . createSavepointDirectory ( root , new JobID ( ) ) ; SavepointV2 savepoint = new SavepointV2 ( checkpointId , CheckpointTestUtils . createOperatorStates ( 4 , 32 ) , Collections . < MasterState > emptyList ( ) ) ; String pathSavepoint = SavepointStore . storeSavepoint ( savepointDirectory2 , savepoint ) ; list = rootFile . listFiles ( ) ; assertNotNull ( list ) ; assertEquals ( 2 , list . length ) ; Savepoint loaded = SavepointStore . loadSavepoint ( pathNewSavepoint , Thread . currentThread ( ) . getContextClassLoader ( ) ) ; assertEquals ( newSavepoint , loaded ) ; loaded = SavepointStore . loadSavepoint ( pathSavepoint , Thread . currentThread ( ) . getContextClassLoader ( ) ) ; assertEquals ( savepoint . getCheckpointId ( ) , loaded . getCheckpointId ( ) ) ; assertEquals ( savepoint . getTaskStates ( ) , loaded . getTaskStates ( ) ) ; assertEquals ( savepoint . getMasterStates ( ) , loaded . getMasterStates ( ) ) ; }
Tests addition of a new savepoint version.
Tests addition of a new savepoint version .
// New savepoint type for test // make this a positive number // make this a positive number // Add serializer // Savepoint v0 // Load