flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackendSnapshotMigrationTest.java
/**
	 * [FLINK-5979]
	 *
	 * This test takes a snapshot that was created with Flink 1.2 and tries to restore it in master to check
	 * the backwards compatibility of the serialization format of {@link StateTable}s.
	 */
[FLINK-5979]
This test takes a snapshot that was created with Flink 1.2 and tries to restore it in master to check
the backwards compatibility of the serialization format of  StateTables.
[ FLINK - 5979 ] This test takes a snapshot that was created with Flink 1 . 2 and tries to restore it in master to check the backwards compatibility of the serialization format of StateTables .
@Test
	public void testRestore1_2ToMaster() throws Exception {
		ClassLoader cl = getClass().getClassLoader();
		URL resource = cl.getResource("heap_keyed_statebackend_1_2.snapshot");
		Preconditions.checkNotNull(resource, "Binary snapshot resource not found!");
		final Integer namespace1 = 1;
		final Integer namespace2 = 2;
		final Integer namespace3 = 3;
		try (final HeapKeyedStateBackend<String> keyedBackend = createKeyedBackend()) {
			final KeyGroupsStateHandle stateHandle;
			try (BufferedInputStream bis = new BufferedInputStream((new FileInputStream(resource.getFile())))) {
				stateHandle = InstantiationUtil.deserializeObject(bis, Thread.currentThread().getContextClassLoader());
			}
			keyedBackend.restore(Collections.<KeyedStateHandle>singleton(stateHandle));
			final ListStateDescriptor<Long> stateDescr = new ListStateDescriptor<>("my-state", Long.class);
			stateDescr.initializeSerializerUnlessSet(new ExecutionConfig());
			InternalListState<Integer, Long> state = keyedBackend.createListState(IntSerializer.INSTANCE, stateDescr);
			assertEquals(7, keyedBackend.numStateEntries());
			keyedBackend.setCurrentKey("abc");
			state.setCurrentNamespace(namespace1);
			assertEquals(asList(33L, 55L), state.get());
			state.setCurrentNamespace(namespace2);
			assertEquals(asList(22L, 11L), state.get());
			state.setCurrentNamespace(namespace3);
			assertEquals(Collections.singletonList(44L), state.get());
			keyedBackend.setCurrentKey("def");
			state.setCurrentNamespace(namespace1);
			assertEquals(asList(11L, 44L), state.get());
			state.setCurrentNamespace(namespace3);
			assertEquals(asList(22L, 55L, 33L), state.get());
			keyedBackend.setCurrentKey("jkl");
			state.setCurrentNamespace(namespace1);
			assertEquals(asList(11L, 22L, 33L, 44L, 55L), state.get());
			keyedBackend.setCurrentKey("mno");
			state.setCurrentNamespace(namespace3);
			assertEquals(asList(11L, 22L, 33L, 44L, 55L), state.get());
		}
	}
@ Test public void testRestore1_2ToMaster ( ) throws Exception { ClassLoader cl = getClass ( ) . getClassLoader ( ) ; URL resource = cl . getResource ( "heap_keyed_statebackend_1_2.snapshot" ) ; Preconditions . checkNotNull ( resource , "Binary snapshot resource not found!" ) ; final Integer namespace1 = 1 ; final Integer namespace2 = 2 ; final Integer namespace3 = 3 ; try ( final HeapKeyedStateBackend < String > keyedBackend = createKeyedBackend ( ) ) { final KeyGroupsStateHandle stateHandle ; try ( BufferedInputStream bis = new BufferedInputStream ( ( new FileInputStream ( resource . getFile ( ) ) ) ) ) { stateHandle = InstantiationUtil . deserializeObject ( bis , Thread . currentThread ( ) . getContextClassLoader ( ) ) ; } keyedBackend . restore ( Collections . < KeyedStateHandle > singleton ( stateHandle ) ) ; final ListStateDescriptor < Long > stateDescr = new ListStateDescriptor < > ( "my-state" , Long . class ) ; stateDescr . initializeSerializerUnlessSet ( new ExecutionConfig ( ) ) ; InternalListState < Integer , Long > state = keyedBackend . createListState ( IntSerializer . INSTANCE , stateDescr ) ; assertEquals ( 7 , keyedBackend . numStateEntries ( ) ) ; keyedBackend . setCurrentKey ( "abc" ) ; state . setCurrentNamespace ( namespace1 ) ; assertEquals ( asList ( 33L , 55L ) , state . get ( ) ) ; state . setCurrentNamespace ( namespace2 ) ; assertEquals ( asList ( 22L , 11L ) , state . get ( ) ) ; state . setCurrentNamespace ( namespace3 ) ; assertEquals ( Collections . singletonList ( 44L ) , state . get ( ) ) ; keyedBackend . setCurrentKey ( "def" ) ; state . setCurrentNamespace ( namespace1 ) ; assertEquals ( asList ( 11L , 44L ) , state . get ( ) ) ; state . setCurrentNamespace ( namespace3 ) ; assertEquals ( asList ( 22L , 55L , 33L ) , state . get ( ) ) ; keyedBackend . setCurrentKey ( "jkl" ) ; state . setCurrentNamespace ( namespace1 ) ; assertEquals ( asList ( 11L , 22L , 33L , 44L , 55L ) , state . get ( ) ) ; keyedBackend . setCurrentKey ( "mno" ) ; state . setCurrentNamespace ( namespace3 ) ; assertEquals ( asList ( 11L , 22L , 33L , 44L , 55L ) , state . get ( ) ) ; } }
[FLINK-5979]
 This test takes a snapshot that was created with Flink 1.2 and tries to restore it in master to check
 the backwards compatibility of the serialization format of {@link StateTable}s.
[ FLINK - 5979 ] This test takes a snapshot that was created with Flink 1 . 2 and tries to restore it in master to check the backwards compatibility of the serialization format of { @link StateTable } s .