OperatorSnapshotResult { public void cancel() throws Exception { Exception exception = null; try { StateUtil.discardStateFuture(getKeyedStateManagedFuture()); } catch (Exception e) { exception = ExceptionUtils.firstOrSuppressed( new Exception("Could not properly cancel managed keyed state future.", e), exception); } try { StateUtil.discardStateFuture(getOperatorStateManagedFuture()); } catch (Exception e) { exception = ExceptionUtils.firstOrSuppressed( new Exception("Could not properly cancel managed operator state future.", e), exception); } try { StateUtil.discardStateFuture(getKeyedStateRawFuture()); } catch (Exception e) { exception = ExceptionUtils.firstOrSuppressed( new Exception("Could not properly cancel raw keyed state future.", e), exception); } try { StateUtil.discardStateFuture(getOperatorStateRawFuture()); } catch (Exception e) { exception = ExceptionUtils.firstOrSuppressed( new Exception("Could not properly cancel raw operator state future.", e), exception); } if (exception != null) { throw exception; } } OperatorSnapshotResult();  OperatorSnapshotResult(
			RunnableFuture<KeyedStateHandle> keyedStateManagedFuture,
			RunnableFuture<KeyedStateHandle> keyedStateRawFuture,
			RunnableFuture<OperatorStateHandle> operatorStateManagedFuture,
			RunnableFuture<OperatorStateHandle> operatorStateRawFuture); RunnableFuture<KeyedStateHandle> getKeyedStateManagedFuture(); void setKeyedStateManagedFuture(RunnableFuture<KeyedStateHandle> keyedStateManagedFuture); RunnableFuture<KeyedStateHandle> getKeyedStateRawFuture(); void setKeyedStateRawFuture(RunnableFuture<KeyedStateHandle> keyedStateRawFuture); RunnableFuture<OperatorStateHandle> getOperatorStateManagedFuture(); void setOperatorStateManagedFuture(RunnableFuture<OperatorStateHandle> operatorStateManagedFuture); RunnableFuture<OperatorStateHandle> getOperatorStateRawFuture(); void setOperatorStateRawFuture(RunnableFuture<OperatorStateHandle> operatorStateRawFuture); void cancel(); boolean hasKeyedState(); boolean hasOperatorState(); boolean hasState();  }
@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(); }