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 registration with the KvStateRegistry.
	 */
Tests registration with the KvStateRegistry.
Tests registration with the KvStateRegistry .
@Test
	public void testQueryableStateRegistration() throws Exception {
		DummyEnvironment env = new DummyEnvironment("test", 1, 0);
		KvStateRegistry registry = env.getKvStateRegistry();
		CheckpointStreamFactory streamFactory = createStreamFactory();
		AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE, env);
		KeyGroupRange expectedKeyGroupRange = backend.getKeyGroupRange();
		KvStateRegistryListener listener = mock(KvStateRegistryListener.class);
		registry.registerListener(listener);
		ValueStateDescriptor<Integer> desc = new ValueStateDescriptor<>(
				"test",
				IntSerializer.INSTANCE);
		desc.setQueryable("banana");
		backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, desc);
		verify(listener, times(1)).notifyKvStateRegistered(
				eq(env.getJobID()), eq(env.getJobVertexId()), eq(expectedKeyGroupRange), eq("banana"), any(KvStateID.class));
		KeyedStateHandle snapshot = FutureUtil.runIfNotDoneAndGet(backend.snapshot(682375462379L, 4, streamFactory, CheckpointOptions.forCheckpoint()));
		backend.dispose();
		verify(listener, times(1)).notifyKvStateUnregistered(
				eq(env.getJobID()), eq(env.getJobVertexId()), eq(expectedKeyGroupRange), eq("banana"));
		backend.dispose();
		backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot, env);
		snapshot.discardState();
		backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, desc);
		 again
		verify(listener, times(2)).notifyKvStateRegistered(
				eq(env.getJobID()), eq(env.getJobVertexId()), eq(expectedKeyGroupRange), eq("banana"), any(KvStateID.class));
		backend.dispose();
	}
@ Test public void testQueryableStateRegistration ( ) throws Exception { DummyEnvironment env = new DummyEnvironment ( "test" , 1 , 0 ) ; KvStateRegistry registry = env . getKvStateRegistry ( ) ; CheckpointStreamFactory streamFactory = createStreamFactory ( ) ; AbstractKeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE , env ) ; KeyGroupRange expectedKeyGroupRange = backend . getKeyGroupRange ( ) ; KvStateRegistryListener listener = mock ( KvStateRegistryListener . class ) ; registry . registerListener ( listener ) ; ValueStateDescriptor < Integer > desc = new ValueStateDescriptor < > ( "test" , IntSerializer . INSTANCE ) ; desc . setQueryable ( "banana" ) ; backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; verify ( listener , times ( 1 ) ) . notifyKvStateRegistered ( eq ( env . getJobID ( ) ) , eq ( env . getJobVertexId ( ) ) , eq ( expectedKeyGroupRange ) , eq ( "banana" ) , any ( KvStateID . class ) ) ; KeyedStateHandle snapshot = FutureUtil . runIfNotDoneAndGet ( backend . snapshot ( 682375462379L , 4 , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ) ; backend . dispose ( ) ; verify ( listener , times ( 1 ) ) . notifyKvStateUnregistered ( eq ( env . getJobID ( ) ) , eq ( env . getJobVertexId ( ) ) , eq ( expectedKeyGroupRange ) , eq ( "banana" ) ) ; backend . dispose ( ) ; backend = restoreKeyedBackend ( IntSerializer . INSTANCE , snapshot , env ) ; snapshot . discardState ( ) ; backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , desc ) ; verify ( listener , times ( 2 ) ) . notifyKvStateRegistered ( eq ( env . getJobID ( ) ) , eq ( env . getJobVertexId ( ) ) , eq ( expectedKeyGroupRange ) , eq ( "banana" ) , any ( KvStateID . class ) ) ; backend . dispose ( ) ; }
Tests registration with the KvStateRegistry.
Tests registration with the KvStateRegistry .
// Verify registered // Initialize again // Verify registered again