public List<StateObject> getDiscardables() {
        return Stream.concat(
                        streamOperatorAndKeyedStates().flatMap(Collection::stream),
                        collectUniqueDelegates(streamChannelStates()))
                .collect(Collectors.toList());
    }
@Test
    void testGetDiscardables() throws IOException {
        Tuple2<List<StateObject>, OperatorSubtaskState> opStates =
                generateSampleOperatorSubtaskState();
        List<StateObject> states = opStates.f0;
        OperatorSubtaskState operatorSubtaskState = opStates.f1;
        List<StateObject> discardables =
                Arrays.asList(
                        states.get(0),
                        states.get(1),
                        states.get(2),
                        states.get(3),
                        ((InputChannelStateHandle) states.get(4)).getDelegate(),
                        ((ResultSubpartitionStateHandle) states.get(5)).getDelegate());
        assertThat(new HashSet<>(operatorSubtaskState.getDiscardables()))
                .isEqualTo(new HashSet<>(discardables));
    }