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

        // check the streamGraph parallelism configured
        final List<StreamNode> streamNodes =
                streamGraph.getStreamNodes().stream()
                        .sorted(Comparator.comparingInt(StreamNode::getId))
                        .collect(Collectors.toList());
        assertThat(streamNodes.get(0).isParallelismConfigured()).isFalse();
        assertThat(streamNodes.get(1).isParallelismConfigured()).isFalse();
        assertThat(streamNodes.get(2).isParallelismConfigured()).isTrue();

        // check the jobGraph parallelism configured
        JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
        List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
        assertThat(jobGraph.getNumberOfVertices()).isEqualTo(1);
        assertThat(vertices.get(0).isParallelismConfigured()).isTrue();
    }