public static Tuple2<Long, Long> discardStateFuture(Future<? extends StateObject> stateFuture)
            throws Exception {
        long stateSize = 0, checkpointedSize = 0;
        if (null != stateFuture) {
            if (!stateFuture.cancel(true)) {

                try {
                    // We attempt to get a result, in case the future completed before cancellation.
                    if (stateFuture instanceof RunnableFuture<?> && !stateFuture.isDone()) {
                        ((RunnableFuture<?>) stateFuture).run();
                    }
                    StateObject stateObject = stateFuture.get();
                    if (stateObject != null) {
                        stateSize = stateObject.getStateSize();
                        checkpointedSize = getCheckpointedSize(stateObject, stateSize);
                        stateObject.discardState();
                    }

                } catch (Exception ex) {
                    LOG.debug(
                            "Cancelled execution of snapshot future runnable. Cancellation produced the following "
                                    + "exception, which is expected an can be ignored.",
                            ex);
                }
            } else if (stateFuture.isDone()) {
                try {
                    StateObject stateObject = stateFuture.get();
                    stateSize = stateObject.getStateSize();
                    checkpointedSize = getCheckpointedSize(stateObject, stateSize);
                } catch (Exception e) {
                    // ignored
                }
            }
        }
        return Tuple2.of(stateSize, checkpointedSize);
    }
@Test
    void testDiscardStateSize() throws Exception {
        assertThat(discardStateFuture(completedFuture(new TestStateObject(1234, 123))))
                .isEqualTo(Tuple2.of(1234L, 123L));
        Tuple2<Long, Long> zeroSize = Tuple2.of(0L, 0L);
        assertThat(discardStateFuture(null)).isEqualTo(zeroSize);
        assertThat(discardStateFuture(new CompletableFuture<>())).isEqualTo(zeroSize);
        assertThat(discardStateFuture(completedExceptionally(new RuntimeException())))
                .isEqualTo(zeroSize);
        assertThat(discardStateFuture(emptyFuture(false, true))).isEqualTo(zeroSize);
        assertThat(discardStateFuture(emptyFuture(false, false))).isEqualTo(zeroSize);
        assertThat(discardStateFuture(emptyFuture(true, true))).isEqualTo(zeroSize);
        assertThat(discardStateFuture(emptyFuture(true, false))).isEqualTo(zeroSize);
    }