public static LineageGraph convertToLineageGraph(List<Transformation<?>> transformations) {
        DefaultLineageGraph.LineageGraphBuilder builder = DefaultLineageGraph.builder();
        for (Transformation<?> transformation : transformations) {
            List<LineageEdge> edges = processSink(transformation);
            for (LineageEdge lineageEdge : edges) {
                builder.addLineageEdge(lineageEdge);
            }
        }
        return builder.build();
    }
@Test
    void testExtractLineageGraphFromLegacyTransformations() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Long> source = env.addSource(new LineageSourceFunction());
        DataStreamSink<Long> sink = source.addSink(new LineageSinkFunction());

        LineageGraph lineageGraph =
                LineageGraphUtils.convertToLineageGraph(Arrays.asList(sink.getTransformation()));

        assertThat(lineageGraph.sources().size()).isEqualTo(1);
        assertThat(lineageGraph.sources().get(0).boundedness())
                .isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
        assertThat(lineageGraph.sources().get(0).datasets().size()).isEqualTo(1);
        assertThat(lineageGraph.sources().get(0).datasets().get(0).name())
                .isEqualTo(LEGACY_SOURCE_DATASET_NAME);
        assertThat(lineageGraph.sources().get(0).datasets().get(0).namespace())
                .isEqualTo(LEGACY_SOURCE_DATASET_NAMESPACE);

        assertThat(lineageGraph.sinks().size()).isEqualTo(1);
        assertThat(lineageGraph.sinks().get(0).datasets().size()).isEqualTo(1);
        assertThat(lineageGraph.sinks().get(0).datasets().get(0).name())
                .isEqualTo(LEGACY_SINK_DATASET_NAME);
        assertThat(lineageGraph.sinks().get(0).datasets().get(0).namespace())
                .isEqualTo(LEGACY_SINK_DATASET_NAMESPACE);

        assertThat(lineageGraph.relations().size()).isEqualTo(1);
    }