flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
/**
	 * The purpose of this test is to check that parallel snapshots are possible, and work even if a previous snapshot
	 * is still running and blocking.
	 */
The purpose of this test is to check that parallel snapshots are possible, and work even if a previous snapshot
is still running and blocking.
The purpose of this test is to check that parallel snapshots are possible and work even if a previous snapshot is still running and blocking .
@Test
	public void testParallelAsyncSnapshots() throws Exception {
		OneShotLatch blocker = new OneShotLatch();
		OneShotLatch waiter = new OneShotLatch();
		BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);
		streamFactory.setWaiterLatch(waiter);
		streamFactory.setBlockerLatch(blocker);
		streamFactory.setAfterNumberInvocations(10);
		final AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE);
		try {
			if (!backend.supportsAsynchronousSnapshots()) {
				return;
			}
			InternalValueState<VoidNamespace, Integer> valueState = backend.createValueState(
				VoidNamespaceSerializer.INSTANCE,
				new ValueStateDescriptor<>("test", IntSerializer.INSTANCE));
			valueState.setCurrentNamespace(VoidNamespace.INSTANCE);
			for (int i = 0; i < 10; ++i) {
				backend.setCurrentKey(i);
				valueState.update(i);
			}
			RunnableFuture<KeyedStateHandle> snapshot1 =
				backend.snapshot(0L, 0L, streamFactory, CheckpointOptions.forCheckpoint());
			Thread runner1 = new Thread(snapshot1, "snapshot-1-runner");
			runner1.start();
			waiter.await();
			for (int i = 5; i < 15; ++i) {
				backend.setCurrentKey(i);
				valueState.update(i + 1);
			}
			streamFactory.setWaiterLatch(null);
			streamFactory.setBlockerLatch(null);
			RunnableFuture<KeyedStateHandle> snapshot2 =
				backend.snapshot(1L, 1L, streamFactory, CheckpointOptions.forCheckpoint());
			Thread runner2 = new Thread(snapshot2,"snapshot-2-runner");
			runner2.start();
			snapshot2.get();
			blocker.trigger();
			snapshot1.get();
		} finally {
			backend.dispose();
		}
	}
@ Test public void testParallelAsyncSnapshots ( ) throws Exception { OneShotLatch blocker = new OneShotLatch ( ) ; OneShotLatch waiter = new OneShotLatch ( ) ; BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory ( 1024 * 1024 ) ; streamFactory . setWaiterLatch ( waiter ) ; streamFactory . setBlockerLatch ( blocker ) ; streamFactory . setAfterNumberInvocations ( 10 ) ; final AbstractKeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE ) ; try { if ( ! backend . supportsAsynchronousSnapshots ( ) ) { return ; } InternalValueState < VoidNamespace , Integer > valueState = backend . createValueState ( VoidNamespaceSerializer . INSTANCE , new ValueStateDescriptor < > ( "test" , IntSerializer . INSTANCE ) ) ; valueState . setCurrentNamespace ( VoidNamespace . INSTANCE ) ; for ( int i = 0 ; i < 10 ; ++ i ) { backend . setCurrentKey ( i ) ; valueState . update ( i ) ; } RunnableFuture < KeyedStateHandle > snapshot1 = backend . snapshot ( 0L , 0L , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ; Thread runner1 = new Thread ( snapshot1 , "snapshot-1-runner" ) ; runner1 . start ( ) ; waiter . await ( ) ; for ( int i = 5 ; i < 15 ; ++ i ) { backend . setCurrentKey ( i ) ; valueState . update ( i + 1 ) ; } streamFactory . setWaiterLatch ( null ) ; streamFactory . setBlockerLatch ( null ) ; RunnableFuture < KeyedStateHandle > snapshot2 = backend . snapshot ( 1L , 1L , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ; Thread runner2 = new Thread ( snapshot2 , "snapshot-2-runner" ) ; runner2 . start ( ) ; snapshot2 . get ( ) ; blocker . trigger ( ) ; snapshot1 . get ( ) ; } finally { backend . dispose ( ) ; } }
The purpose of this test is to check that parallel snapshots are possible, and work even if a previous snapshot
 is still running and blocking.
The purpose of this test is to check that parallel snapshots are possible and work even if a previous snapshot is still running and blocking .
// insert some data to the backend. // after this call returns, we have a running snapshot-1 that is blocked in IO. // do some updates in between the snapshots. // we don't want to block the second snapshot. // snapshot-2 should run and succeed, while snapshot-1 is still running and blocked in IO. // we release the blocking IO so that snapshot-1 can also finish and succeed.