@VisibleForTesting
    public static JobGraph createJobGraph(StreamGraph streamGraph) {
        return new StreamingJobGraphGenerator(
                        Thread.currentThread().getContextClassLoader(),
                        streamGraph,
                        null,
                        Runnable::run)
                .createJobGraph();
    }
@Test
    void testDynamicGraphVertexParallelism() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        int defaultParallelism = 20;
        env.setParallelism(defaultParallelism);
        env.fromSequence(1L, 3L).map(value -> value).print();
        StreamGraph streamGraph = env.getStreamGraph();

        for (StreamNode streamNode : streamGraph.getStreamNodes()) {
            assertThat(streamNode.getParallelism()).isEqualTo(defaultParallelism);
        }
        streamGraph.setDynamic(false);
        JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
        List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
        for (JobVertex vertex : vertices) {
            assertThat(vertex.getParallelism()).isEqualTo(defaultParallelism);
        }

        for (StreamNode streamNode : streamGraph.getStreamNodes()) {
            assertThat(streamNode.getParallelism()).isEqualTo(defaultParallelism);
        }
        streamGraph.setDynamic(true);
        jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
        vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
        for (JobVertex vertex : vertices) {
            assertThat(vertex.getParallelism()).isEqualTo(ExecutionConfig.PARALLELISM_DEFAULT);
        }
    }