@Override
    public long getStateSize() {
        return streamOperatorSubtaskStates().mapToLong(StateObject::getStateSize).sum();
    }
@Test
    void getStateSize() {
        Random random = new Random(0x42);
        TaskStateSnapshot taskStateSnapshot = new TaskStateSnapshot();
        assertThat(taskStateSnapshot.getStateSize()).isZero();

        OperatorSubtaskState emptyOperatorSubtaskState = OperatorSubtaskState.builder().build();
        assertThat(emptyOperatorSubtaskState.hasState()).isFalse();
        taskStateSnapshot.putSubtaskStateByOperatorID(new OperatorID(), emptyOperatorSubtaskState);
        assertThat(taskStateSnapshot.getStateSize()).isZero();

        OperatorStateHandle stateHandle_1 =
                StateHandleDummyUtil.createNewOperatorStateHandle(2, random);
        OperatorSubtaskState nonEmptyOperatorSubtaskState_1 =
                OperatorSubtaskState.builder().setManagedOperatorState(stateHandle_1).build();

        OperatorStateHandle stateHandle_2 =
                StateHandleDummyUtil.createNewOperatorStateHandle(2, random);
        OperatorSubtaskState nonEmptyOperatorSubtaskState_2 =
                OperatorSubtaskState.builder().setRawOperatorState(stateHandle_2).build();

        taskStateSnapshot.putSubtaskStateByOperatorID(
                new OperatorID(), nonEmptyOperatorSubtaskState_1);
        taskStateSnapshot.putSubtaskStateByOperatorID(
                new OperatorID(), nonEmptyOperatorSubtaskState_2);

        long totalSize = stateHandle_1.getStateSize() + stateHandle_2.getStateSize();
        assertThat(taskStateSnapshot.getStateSize()).isEqualTo(totalSize);
    }