@Internal
    public StreamGraph getStreamGraph() {
        return getStreamGraph(true);
    }
@Test
    void testGetStreamGraph() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Integer> dataStream1 = env.fromData(1, 2, 3);
        dataStream1.sinkTo(new DiscardingSink<>());
        assertThat(env.getStreamGraph().getStreamNodes().size()).isEqualTo(2);

        DataStreamSource<Integer> dataStream2 = env.fromData(1, 2, 3);
        dataStream2.sinkTo(new DiscardingSink<>());
        // Previous getStreamGraph() call cleaned dataStream1 transformations
        assertThat(env.getStreamGraph().getStreamNodes().size()).isEqualTo(2);

        DataStreamSource<Integer> dataStream3 = env.fromData(1, 2, 3);
        dataStream3.sinkTo(new DiscardingSink<>());
        // Does not clear the transformations.
        env.getExecutionPlan();
        DataStreamSource<Integer> dataStream4 = env.fromData(1, 2, 3);
        dataStream4.sinkTo(new DiscardingSink<>());
        // dataStream3 are preserved
        assertThat(env.getStreamGraph().getStreamNodes().size()).isEqualTo(4);
    }