@Override
    public void discardState() throws Exception {
        StateUtil.bestEffortDiscardAllStateObjects(subtaskStatesByOperatorID.values());
    }
@Test
    void discardState() throws Exception {
        TaskStateSnapshot taskStateSnapshot = new TaskStateSnapshot();
        OperatorID operatorID_1 = new OperatorID();
        OperatorID operatorID_2 = new OperatorID();

        OperatorSubtaskState operatorSubtaskState_1 = mock(OperatorSubtaskState.class);
        OperatorSubtaskState operatorSubtaskState_2 = mock(OperatorSubtaskState.class);

        taskStateSnapshot.putSubtaskStateByOperatorID(operatorID_1, operatorSubtaskState_1);
        taskStateSnapshot.putSubtaskStateByOperatorID(operatorID_2, operatorSubtaskState_2);

        taskStateSnapshot.discardState();
        verify(operatorSubtaskState_1).discardState();
        verify(operatorSubtaskState_2).discardState();
    }