flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
/**
	 * Tests that {@link AbstractHeapState} instances respect the queryable
	 * flag and create concurrent variants for internal state structures.
	 */
Tests that  AbstractHeapState instances respect the queryable
flag and create concurrent variants for internal state structures.
Tests that AbstractHeapState instances respect the queryable flag and create concurrent variants for internal state structures .
@SuppressWarnings("unchecked")
	protected void testConcurrentMapIfQueryable() throws Exception {
		final int numberOfKeyGroups = 1;
		final AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(
				IntSerializer.INSTANCE,
				numberOfKeyGroups,
				new KeyGroupRange(0, 0),
				new DummyEnvironment("test_op", 1, 0));
		{
			ValueStateDescriptor<Integer> desc = new ValueStateDescriptor<>(
					"value-state",
					Integer.class,
					-1);
			desc.setQueryable("my-query");
			ValueState<Integer> state = backend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					desc);
			InternalKvState<VoidNamespace> kvState = (InternalKvState<VoidNamespace>) state;
			assertTrue(kvState instanceof AbstractHeapState);
			kvState.setCurrentNamespace(VoidNamespace.INSTANCE);
			backend.setCurrentKey(1);
			state.update(121818273);
			StateTable<?, ?, ?> stateTable = ((AbstractHeapState<?, ?,? ,?, ?>) kvState).getStateTable();
			checkConcurrentStateTable(stateTable, numberOfKeyGroups);
		}
		{
			ListStateDescriptor<Integer> desc = new ListStateDescriptor<>("list-state", Integer.class);
			desc.setQueryable("my-query");
			ListState<Integer> state = backend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					desc);
			InternalKvState<VoidNamespace> kvState = (InternalKvState<VoidNamespace>) state;
			assertTrue(kvState instanceof AbstractHeapState);
			kvState.setCurrentNamespace(VoidNamespace.INSTANCE);
			backend.setCurrentKey(1);
			state.add(121818273);
			StateTable<?, ?, ?> stateTable = ((AbstractHeapState<?, ?,? ,?, ?>) kvState).getStateTable();
			checkConcurrentStateTable(stateTable, numberOfKeyGroups);
		}
		{
			ReducingStateDescriptor<Integer> desc = new ReducingStateDescriptor<>(
					"reducing-state", new ReduceFunction<Integer>() {
				@Override
				public Integer reduce(Integer value1, Integer value2) throws Exception {
					return value1 + value2;
				}
			}, Integer.class);
			desc.setQueryable("my-query");
			ReducingState<Integer> state = backend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					desc);
			InternalKvState<VoidNamespace> kvState = (InternalKvState<VoidNamespace>) state;
			assertTrue(kvState instanceof AbstractHeapState);
			kvState.setCurrentNamespace(VoidNamespace.INSTANCE);
			backend.setCurrentKey(1);
			state.add(121818273);
			StateTable<?, ?, ?> stateTable = ((AbstractHeapState<?, ?,? ,?, ?>) kvState).getStateTable();
			checkConcurrentStateTable(stateTable, numberOfKeyGroups);
		}
		{
			FoldingStateDescriptor<Integer, Integer> desc = new FoldingStateDescriptor<>(
					"folding-state", 0, new FoldFunction<Integer, Integer>() {
				@Override
				public Integer fold(Integer accumulator, Integer value) throws Exception {
					return accumulator + value;
				}
			}, Integer.class);
			desc.setQueryable("my-query");
			FoldingState<Integer, Integer> state = backend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					desc);
			InternalKvState<VoidNamespace> kvState = (InternalKvState<VoidNamespace>) state;
			assertTrue(kvState instanceof AbstractHeapState);
			kvState.setCurrentNamespace(VoidNamespace.INSTANCE);
			backend.setCurrentKey(1);
			state.add(121818273);
			StateTable<?, ?, ?> stateTable = ((AbstractHeapState<?, ?,? ,?, ?>) kvState).getStateTable();
			checkConcurrentStateTable(stateTable, numberOfKeyGroups);
		}
		{
			MapStateDescriptor<Integer, String> desc = new MapStateDescriptor<>("map-state", Integer.class, String.class);
			desc.setQueryable("my-query");
			MapState<Integer, String> state = backend.getPartitionedState(
					VoidNamespace.INSTANCE,
					VoidNamespaceSerializer.INSTANCE,
					desc);
			InternalKvState<VoidNamespace> kvState = (InternalKvState<VoidNamespace>) state;
			assertTrue(kvState instanceof AbstractHeapState);
			kvState.setCurrentNamespace(VoidNamespace.INSTANCE);
			backend.setCurrentKey(1);
			state.put(121818273, "121818273");
			int keyGroupIndex = KeyGroupRangeAssignment.assignToKeyGroup(1, numberOfKeyGroups);
			StateTable stateTable = ((AbstractHeapState) kvState).getStateTable();
			assertNotNull("State not set", stateTable.get(keyGroupIndex));
			checkConcurrentStateTable(stateTable, numberOfKeyGroups);
		}
		backend.dispose();
	}
@ SuppressWarnings ( "unchecked" ) protected void testConcurrentMapIfQueryable ( ) throws Exception { final int numberOfKeyGroups = 1 ; final AbstractKeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE , numberOfKeyGroups , new KeyGroupRange ( 0 , 0 ) , new DummyEnvironment ( "test_op" , 1 , 0 ) ) ; { ValueStateDescriptor < Integer > desc = new ValueStateDescriptor < > ( "value-state" , Integer . class , - 1 ) ; desc . setQueryable ( "my-query" ) ; ValueState < Integer > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; InternalKvState < VoidNamespace > kvState = ( InternalKvState < VoidNamespace > ) state ; assertTrue ( kvState instanceof AbstractHeapState ) ; kvState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; backend . setCurrentKey ( 1 ) ; state . update ( 121818273 ) ; StateTable < ? , ? , ? > stateTable = ( ( AbstractHeapState < ? , ? , ? , ? , ? > ) kvState ) . getStateTable ( ) ; checkConcurrentStateTable ( stateTable , numberOfKeyGroups ) ; } { ListStateDescriptor < Integer > desc = new ListStateDescriptor < > ( "list-state" , Integer . class ) ; desc . setQueryable ( "my-query" ) ; ListState < Integer > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; InternalKvState < VoidNamespace > kvState = ( InternalKvState < VoidNamespace > ) state ; assertTrue ( kvState instanceof AbstractHeapState ) ; kvState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; backend . setCurrentKey ( 1 ) ; state . add ( 121818273 ) ; StateTable < ? , ? , ? > stateTable = ( ( AbstractHeapState < ? , ? , ? , ? , ? > ) kvState ) . getStateTable ( ) ; checkConcurrentStateTable ( stateTable , numberOfKeyGroups ) ; } { ReducingStateDescriptor < Integer > desc = new ReducingStateDescriptor < > ( "reducing-state" , new ReduceFunction < Integer > ( ) { @ Override public Integer reduce ( Integer value1 , Integer value2 ) throws Exception { return value1 + value2 ; } } , Integer . class ) ; desc . setQueryable ( "my-query" ) ; ReducingState < Integer > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; InternalKvState < VoidNamespace > kvState = ( InternalKvState < VoidNamespace > ) state ; assertTrue ( kvState instanceof AbstractHeapState ) ; kvState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; backend . setCurrentKey ( 1 ) ; state . add ( 121818273 ) ; StateTable < ? , ? , ? > stateTable = ( ( AbstractHeapState < ? , ? , ? , ? , ? > ) kvState ) . getStateTable ( ) ; checkConcurrentStateTable ( stateTable , numberOfKeyGroups ) ; } { FoldingStateDescriptor < Integer , Integer > desc = new FoldingStateDescriptor < > ( "folding-state" , 0 , new FoldFunction < Integer , Integer > ( ) { @ Override public Integer fold ( Integer accumulator , Integer value ) throws Exception { return accumulator + value ; } } , Integer . class ) ; desc . setQueryable ( "my-query" ) ; FoldingState < Integer , Integer > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; InternalKvState < VoidNamespace > kvState = ( InternalKvState < VoidNamespace > ) state ; assertTrue ( kvState instanceof AbstractHeapState ) ; kvState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; backend . setCurrentKey ( 1 ) ; state . add ( 121818273 ) ; StateTable < ? , ? , ? > stateTable = ( ( AbstractHeapState < ? , ? , ? , ? , ? > ) kvState ) . getStateTable ( ) ; checkConcurrentStateTable ( stateTable , numberOfKeyGroups ) ; } { MapStateDescriptor < Integer , String > desc = new MapStateDescriptor < > ( "map-state" , Integer . class , String . class ) ; desc . setQueryable ( "my-query" ) ; MapState < Integer , String > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; InternalKvState < VoidNamespace > kvState = ( InternalKvState < VoidNamespace > ) state ; assertTrue ( kvState instanceof AbstractHeapState ) ; kvState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; backend . setCurrentKey ( 1 ) ; state . put ( 121818273 , "121818273" ) ; int keyGroupIndex = KeyGroupRangeAssignment . assignToKeyGroup ( 1 , numberOfKeyGroups ) ; StateTable stateTable = ( ( AbstractHeapState ) kvState ) . getStateTable ( ) ; assertNotNull ( "State not set" , stateTable . get ( keyGroupIndex ) ) ; checkConcurrentStateTable ( stateTable , numberOfKeyGroups ) ; } backend . dispose ( ) ; }
Tests that {@link AbstractHeapState} instances respect the queryable
 flag and create concurrent variants for internal state structures.
Tests that { @link AbstractHeapState } instances respect the queryable flag and create concurrent variants for internal state structures .
// ValueState // ListState // ReducingState // FoldingState // MapState