@Override
    public void setParallelism(int parallelism) {
        checkParallelism(parallelism);
        Preconditions.checkArgument(
                parallelism <= maxParallelism,
                "Vertex's parallelism should be smaller than or equal to vertex's max parallelism.");
        Preconditions.checkState(
                this.parallelism == ExecutionConfig.PARALLELISM_DEFAULT,
                "Vertex's parallelism can be set only if the vertex's parallelism was not decided yet.");
        this.parallelism = parallelism;
    }
@Test
    void testSetParallelism() {
        DefaultVertexParallelismInfo info =
                new DefaultVertexParallelismInfo(
                        ExecutionConfig.PARALLELISM_DEFAULT, 10, ALWAYS_VALID);

        // test set negative value
        assertThatThrownBy(() -> info.setParallelism(-1))
                .withFailMessage("parallelism is not in valid bounds")
                .isInstanceOf(IllegalArgumentException.class);

        // test parallelism larger than max parallelism
        assertThatThrownBy(() -> info.setParallelism(11))
                .withFailMessage(
                        "Vertex's parallelism should be smaller than or equal to vertex's max parallelism.")
                .isInstanceOf(IllegalArgumentException.class);

        // set valid value.
        info.setParallelism(5);

        // test set parallelism for vertex whose parallelism was decided.
        assertThatThrownBy(() -> info.setParallelism(5))
                .withFailMessage(
                        "Vertex's parallelism can be set only if the vertex's parallelism was not decided yet.")
                .isInstanceOf(IllegalStateException.class);
    }