public Tuple2<Long, Long> cancel() throws Exception {
        List<Tuple2<Future<? extends StateObject>, String>> pairs = new ArrayList<>();
        pairs.add(new Tuple2<>(getKeyedStateManagedFuture(), "managed keyed"));
        pairs.add(new Tuple2<>(getKeyedStateRawFuture(), "managed operator"));
        pairs.add(new Tuple2<>(getOperatorStateManagedFuture(), "raw keyed"));
        pairs.add(new Tuple2<>(getOperatorStateRawFuture(), "raw operator"));
        pairs.add(new Tuple2<>(getInputChannelStateFuture(), "input channel"));
        pairs.add(new Tuple2<>(getResultSubpartitionStateFuture(), "result subpartition"));
        final long[] sizeTuple = new long[2];
        try (Closer closer = Closer.create()) {
            for (Tuple2<Future<? extends StateObject>, String> pair : pairs) {
                closer.register(
                        () -> {
                            try {
                                Tuple2<Long, Long> tuple = discardStateFuture(pair.f0);
                                sizeTuple[0] += tuple.f0;
                                sizeTuple[1] += tuple.f1;
                            } catch (Exception e) {
                                throw new RuntimeException(
                                        String.format(
                                                "Could not properly cancel %s state future",
                                                pair.f1),
                                        e);
                            }
                        });
            }
        }
        return Tuple2.of(sizeTuple[0], sizeTuple[1]);
    }
@Test
    void testCancelAndCleanup() throws Exception {
        OperatorSnapshotFutures operatorSnapshotResult = new OperatorSnapshotFutures();

        operatorSnapshotResult.cancel();

        KeyedStateHandle keyedManagedStateHandle = mock(KeyedStateHandle.class);
        SnapshotResult<KeyedStateHandle> keyedStateManagedResult =
                SnapshotResult.of(keyedManagedStateHandle);
        RunnableFuture<SnapshotResult<KeyedStateHandle>> keyedStateManagedFuture =
                spy(DoneFuture.of(keyedStateManagedResult));

        KeyedStateHandle keyedRawStateHandle = mock(KeyedStateHandle.class);
        SnapshotResult<KeyedStateHandle> keyedStateRawResult =
                SnapshotResult.of(keyedRawStateHandle);
        RunnableFuture<SnapshotResult<KeyedStateHandle>> keyedStateRawFuture =
                spy(DoneFuture.of(keyedStateRawResult));

        OperatorStateHandle operatorManagedStateHandle = mock(OperatorStreamStateHandle.class);
        SnapshotResult<OperatorStateHandle> operatorStateManagedResult =
                SnapshotResult.of(operatorManagedStateHandle);
        RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateManagedFuture =
                spy(DoneFuture.of(operatorStateManagedResult));

        OperatorStateHandle operatorRawStateHandle = mock(OperatorStreamStateHandle.class);
        SnapshotResult<OperatorStateHandle> operatorStateRawResult =
                SnapshotResult.of(operatorRawStateHandle);
        RunnableFuture<SnapshotResult<OperatorStateHandle>> operatorStateRawFuture =
                spy(DoneFuture.of(operatorStateRawResult));

        InputChannelStateHandle inputChannelRawStateHandle = mock(InputChannelStateHandle.class);
        SnapshotResult<StateObjectCollection<InputChannelStateHandle>> inputChannelStateRawResult =
                SnapshotResult.of(StateObjectCollection.singleton(inputChannelRawStateHandle));
        Future<SnapshotResult<StateObjectCollection<InputChannelStateHandle>>>
                inputChannelStateRawFuture = spy(DoneFuture.of(inputChannelStateRawResult));

        ResultSubpartitionStateHandle resultSubpartitionRawStateHandle =
                mock(ResultSubpartitionStateHandle.class);
        SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>
                resultSubpartitionStateRawResult =
                        SnapshotResult.of(
                                StateObjectCollection.singleton(resultSubpartitionRawStateHandle));
        Future<SnapshotResult<StateObjectCollection<ResultSubpartitionStateHandle>>>
                resultSubpartitionStateRawFuture =
                        spy(DoneFuture.of(resultSubpartitionStateRawResult));

        operatorSnapshotResult =
                new OperatorSnapshotFutures(
                        keyedStateManagedFuture,
                        keyedStateRawFuture,
                        operatorStateManagedFuture,
                        operatorStateRawFuture,
                        inputChannelStateRawFuture,
                        resultSubpartitionStateRawFuture);

        operatorSnapshotResult.cancel();

        verify(keyedStateManagedFuture).cancel(true);
        verify(keyedStateRawFuture).cancel(true);
        verify(operatorStateManagedFuture).cancel(true);
        verify(operatorStateRawFuture).cancel(true);
        verify(inputChannelStateRawFuture).cancel(true);
        verify(resultSubpartitionStateRawFuture).cancel(true);

        verify(keyedManagedStateHandle).discardState();
        verify(keyedRawStateHandle).discardState();
        verify(operatorManagedStateHandle).discardState();
        verify(operatorRawStateHandle).discardState();
        verify(inputChannelRawStateHandle).discardState();
        verify(resultSubpartitionRawStateHandle).discardState();
    }