@VisibleForTesting
    public static JobGraph createJobGraph(StreamGraph streamGraph) {
        return new StreamingJobGraphGenerator(
                        Thread.currentThread().getContextClassLoader(),
                        streamGraph,
                        null,
                        Runnable::run)
                .createJobGraph();
    }
@Test
    void testIntermediateDataSetReuse() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setBufferTimeout(-1);
        DataStream<Integer> source = env.fromData(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);

        // these two vertices can reuse the same intermediate dataset
        source.rebalance().sinkTo(new DiscardingSink<>()).setParallelism(2).name("sink1");
        source.rebalance().sinkTo(new DiscardingSink<>()).setParallelism(2).name("sink2");

        // this can not reuse the same intermediate dataset because of different parallelism
        source.rebalance().sinkTo(new DiscardingSink<>()).setParallelism(3);

        // this can not reuse the same intermediate dataset because of different partitioner
        source.broadcast().sinkTo(new DiscardingSink<>()).setParallelism(2);

        // these two vertices can not reuse the same intermediate dataset because of the pipelined
        // edge
        source.forward().sinkTo(new DiscardingSink<>()).setParallelism(1).disableChaining();
        source.forward().sinkTo(new DiscardingSink<>()).setParallelism(1).disableChaining();

        DataStream<Integer> mapStream = source.forward().map(value -> value).setParallelism(1);

        // these two vertices can reuse the same intermediate dataset
        mapStream.broadcast().sinkTo(new DiscardingSink<>()).setParallelism(2).name("sink3");
        mapStream.broadcast().sinkTo(new DiscardingSink<>()).setParallelism(2).name("sink4");

        StreamGraph streamGraph = env.getStreamGraph();
        streamGraph.setGlobalStreamExchangeMode(GlobalStreamExchangeMode.FORWARD_EDGES_PIPELINED);
        JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);

        List<JobVertex> vertices = jobGraph.getVerticesSortedTopologicallyFromSources();
        assertThat(vertices).hasSize(9);

        JobVertex sourceVertex = vertices.get(0);
        List<IntermediateDataSetID> producedDataSet =
                sourceVertex.getProducedDataSets().stream()
                        .map(IntermediateDataSet::getId)
                        .collect(Collectors.toList());
        assertThat(producedDataSet).hasSize(6);

        JobVertex sinkVertex1 = checkNotNull(findJobVertexWithName(vertices, "sink1"));
        JobVertex sinkVertex2 = checkNotNull(findJobVertexWithName(vertices, "sink2"));
        JobVertex sinkVertex3 = checkNotNull(findJobVertexWithName(vertices, "sink3"));
        JobVertex sinkVertex4 = checkNotNull(findJobVertexWithName(vertices, "sink4"));

        assertThat(sinkVertex2.getInputs().get(0).getSource().getId())
                .isEqualTo(sinkVertex1.getInputs().get(0).getSource().getId());
        assertThat(sinkVertex4.getInputs().get(0).getSource().getId())
                .isEqualTo(sinkVertex3.getInputs().get(0).getSource().getId());
        assertThat(sinkVertex3.getInputs().get(0).getSource().getId())
                .isNotEqualTo(sinkVertex1.getInputs().get(0).getSource().getId());

        StreamConfig streamConfig = new StreamConfig(sourceVertex.getConfiguration());
        List<IntermediateDataSetID> nonChainedOutputs =
                streamConfig.getOperatorNonChainedOutputs(getClass().getClassLoader()).stream()
                        .map(NonChainedOutput::getDataSetId)
                        .collect(Collectors.toList());
        assertThat(nonChainedOutputs).hasSize(5);
        assertThat(nonChainedOutputs)
                .doesNotContain(sinkVertex3.getInputs().get(0).getSource().getId());

        List<IntermediateDataSetID> streamOutputsInOrder =
                streamConfig.getVertexNonChainedOutputs(getClass().getClassLoader()).stream()
                        .map(NonChainedOutput::getDataSetId)
                        .collect(Collectors.toList());
        assertThat(streamOutputsInOrder).hasSize(6);
        assertThat(streamOutputsInOrder).isEqualTo(producedDataSet);
    }