import java.util.concurrent.RunnableFuture; //导入方法依赖的package包/类

@Test

public void testSnapshotAsyncCancel() throws Exception {

DefaultOperatorStateBackend operatorStateBackend =

new DefaultOperatorStateBackend(OperatorStateBackendTest.class.getClassLoader(), new ExecutionConfig(), true);

ListStateDescriptor stateDescriptor1 =

new ListStateDescriptor<>("test1", new JavaSerializer());

ListState listState1 = operatorStateBackend.getOperatorState(stateDescriptor1);

listState1.add(MutableType.of(42));

listState1.add(MutableType.of(4711));

BlockerCheckpointStreamFactory streamFactory = new BlockerCheckpointStreamFactory(1024 * 1024);

OneShotLatch waiterLatch = new OneShotLatch();

OneShotLatch blockerLatch = new OneShotLatch();

streamFactory.setWaiterLatch(waiterLatch);

streamFactory.setBlockerLatch(blockerLatch);

RunnableFuture runnableFuture =

operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forCheckpoint());

ExecutorService executorService = Executors.newFixedThreadPool(1);

executorService.submit(runnableFuture);

// wait until the async checkpoint is in the stream's write code, then continue

waiterLatch.await();

// cancel the future, which should close the underlying stream

runnableFuture.cancel(true);

Assert.assertTrue(streamFactory.getLastCreatedStream().isClosed());

// we allow the stream under test to proceed

blockerLatch.trigger();

try {

runnableFuture.get(60, TimeUnit.SECONDS);

Assert.fail();

} catch (CancellationException ignore) {

}

}