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 testCancelReturnsStateSize() throws Exception {
        KeyGroupsStateHandle s1 =
                new KeyGroupsStateHandle(
                        new KeyGroupRangeOffsets(0, 0),
                        new ByteStreamStateHandle("", new byte[123]));
        KeyGroupsStateHandle s2 =
                new KeyGroupsStateHandle(
                        new KeyGroupRangeOffsets(0, 0),
                        new ByteStreamStateHandle("", new byte[456]));
        OperatorSnapshotFutures futures =
                new OperatorSnapshotFutures(
                        DoneFuture.of(SnapshotResult.of(s1)),
                        DoneFuture.of(SnapshotResult.of(s2)),
                        DoneFuture.of(SnapshotResult.empty()),
                        ExceptionallyDoneFuture.of(new RuntimeException()),
                        ExceptionallyDoneFuture.of(new RuntimeException()),
                        ExceptionallyDoneFuture.of(new RuntimeException()));
        long stateSize = s1.getStateSize() + s2.getStateSize();
        assertThat(futures.cancel()).isEqualTo(Tuple2.of(stateSize, stateSize));
    }