@Override
    public <OUT> NonKeyedPartitionStream<OUT> fromSource(Source<OUT> source, String sourceName) {
        if (source instanceof WrappedSource) {
            org.apache.flink.api.connector.source.Source<OUT, ?, ?> innerSource =
                    ((WrappedSource<OUT>) source).getWrappedSource();
            final TypeInformation<OUT> resolvedTypeInfo =
                    getSourceTypeInfo(innerSource, sourceName);

            SourceTransformation<OUT, ?, ?> sourceTransformation =
                    new SourceTransformation<>(
                            sourceName,
                            innerSource,
                            WatermarkStrategy.noWatermarks(),
                            resolvedTypeInfo,
                            getParallelism(),
                            false);
            return new NonKeyedPartitionStreamImpl<>(this, sourceTransformation);
        } else if (source instanceof FromDataSource) {
            Collection<OUT> data = ((FromDataSource<OUT>) source).getData();
            TypeInformation<OUT> outType = extractTypeInfoFromCollection(data);

            FromElementsGeneratorFunction<OUT> generatorFunction =
                    new FromElementsGeneratorFunction<>(outType, executionConfig, data);

            DataGeneratorSource<OUT> generatorSource =
                    new DataGeneratorSource<>(generatorFunction, data.size(), outType);

            return fromSource(new WrappedSource<>(generatorSource), "Collection Source");
        } else {
            throw new UnsupportedOperationException(
                    "Unsupported type of sink, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source.");
        }
    }
@Test
    void testFromSource() {
        ExecutionEnvironmentImpl env =
                new ExecutionEnvironmentImpl(
                        new DefaultExecutorServiceLoader(), new Configuration(), null);
        NonKeyedPartitionStream<Integer> source =
                env.fromSource(
                        DataStreamV2SourceUtils.fromData(Arrays.asList(1, 2, 3)), "test-source");
        source.process(new StreamTestUtils.NoOpOneInputStreamProcessFunction());
        StreamGraph streamGraph = StreamTestUtils.getStreamGraph(env);
        Collection<StreamNode> nodes = streamGraph.getStreamNodes();
        assertThat(nodes).hasSize(2);
        Collection<Integer> sourceIDs = streamGraph.getSourceIDs();
        for (StreamNode node : nodes) {
            if (node.getOperatorName().contains("source")) {
                assertThat(sourceIDs).containsExactly(node.getId());
            }
        }
    }