@Test
public void testCancelAndCleanup() throws Exception {

	OperatorSnapshotResult operatorSnapshotResult = new OperatorSnapshotResult();
	operatorSnapshotResult.cancel();
	KeyedStateHandle keyedManagedStateHandle = mock(KeyedStateHandle.class);
	RunnableFuture<KeyedStateHandle> keyedStateManagedFuture = mock(RunnableFuture.class);
	when(keyedStateManagedFuture.get()).thenReturn(keyedManagedStateHandle);
	KeyedStateHandle keyedRawStateHandle = mock(KeyedStateHandle.class);
	RunnableFuture<KeyedStateHandle> keyedStateRawFuture = mock(RunnableFuture.class);
	when(keyedStateRawFuture.get()).thenReturn(keyedRawStateHandle);
	OperatorStateHandle operatorManagedStateHandle = mock(OperatorStateHandle.class);
	RunnableFuture<OperatorStateHandle> operatorStateManagedFuture = mock(RunnableFuture.class);
	when(operatorStateManagedFuture.get()).thenReturn(operatorManagedStateHandle);
	OperatorStateHandle operatorRawStateHandle = mock(OperatorStateHandle.class);
	RunnableFuture<OperatorStateHandle> operatorStateRawFuture = mock(RunnableFuture.class);
	when(operatorStateRawFuture.get()).thenReturn(operatorRawStateHandle);
	operatorSnapshotResult = new OperatorSnapshotResult(
		keyedStateManagedFuture,
		keyedStateRawFuture,
		operatorStateManagedFuture,
		operatorStateRawFuture);
	operatorSnapshotResult.cancel();
	verify(keyedStateManagedFuture).cancel(true);
	verify(keyedStateRawFuture).cancel(true);
	verify(operatorStateManagedFuture).cancel(true);
	verify(operatorStateRawFuture).cancel(true);
	verify(keyedManagedStateHandle).discardState();
	verify(keyedRawStateHandle).discardState();
	verify(operatorManagedStateHandle).discardState();
	verify(operatorRawStateHandle).discardState();
}