public FullyFinishedOperatorState(OperatorID operatorID, int parallelism, int maxParallelism) {
        super(operatorID, parallelism, maxParallelism);
    }
@Test
    void testFullyFinishedOperatorState() {
        OperatorState operatorState = new FullyFinishedOperatorState(new OperatorID(), 5, 256);
        assertThat(operatorState.isFullyFinished()).isTrue();

        assertThat(operatorState.getSubtaskStates()).isEmpty();
        assertThat(operatorState.getStates()).isEmpty();
        assertThat(operatorState.getNumberCollectedStates()).isZero();

        assertThatThrownBy(() -> operatorState.putState(0, OperatorSubtaskState.builder().build()))
                .as("Should not be able to put new subtask states for a fully finished state")
                .isInstanceOf(UnsupportedOperationException.class);

        assertThatThrownBy(
                        () ->
                                operatorState.setCoordinatorState(
                                        new ByteStreamStateHandle("test", new byte[] {1, 2, 3, 4})))
                .as("Should not be able to put new subtask states for a fully finished state")
                .isInstanceOf(UnsupportedOperationException.class);
    }