{"record":{"id":"ed4a2048a6904a5a","repo":"apache/flink","slug":"unsupported-type-of-source-you-could-use-datastre","errorCode":null,"errorMessage":"Unsupported type of source, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source.","messagePattern":"Unsupported type of source, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java","lineNumber":195,"sourceCode":"                            WatermarkStrategy.noWatermarks(),\n                            resolvedTypeInfo,\n                            getParallelism(),\n                            false);\n            return StreamUtils.wrapWithConfigureHandle(\n                    new NonKeyedPartitionStreamImpl<>(this, sourceTransformation));\n        } else if (source instanceof FromDataSource) {\n            Collection<OUT> data = ((FromDataSource<OUT>) source).getData();\n            TypeInformation<OUT> outType = extractTypeInfoFromCollection(data);\n\n            FromElementsGeneratorFunction<OUT> generatorFunction =\n                    new FromElementsGeneratorFunction<>(outType, executionConfig, data);\n\n            DataGeneratorSource<OUT> generatorSource =\n                    new DataGeneratorSource<>(generatorFunction, data.size(), outType);\n\n            return fromSource(new WrappedSource<>(generatorSource), \"Collection Source\");\n        } else {\n            throw new UnsupportedOperationException(\n                    \"Unsupported type of source, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source.\");\n        }\n    }\n\n    public Configuration getConfiguration() {\n        return this.configuration;\n    }\n\n    public ExecutionConfig getExecutionConfig() {\n        return executionConfig;\n    }\n\n    public int getParallelism() {\n        return executionConfig.getParallelism();\n    }\n\n    public List<Transformation<?>> getTransformations() {\n        return transformations;","sourceCodeStart":177,"sourceCodeEnd":213,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java#L177-L213","documentation":"ExecutionEnvironmentImpl.fromSource throws UnsupportedOperationException when the passed source is neither a supported internal source type (e.g. WrappedSource/FLIP-27 based) nor a FromDataSource collection source. The DataStream v2 environment only accepts sources that expose a FLIP-27 Source; arbitrary SourceFunction/legacy or custom source objects are rejected with a hint to use DataStreamV2SourceUtils.","triggerScenarios":"Passing a legacy SourceFunction-based connector, an arbitrary object implementing some other source interface, or a FLINK-27 Source not wrapped for v2, directly to env.fromSource(...).","commonSituations":"Porting DataStream v1 jobs: reusing old connector instances or a KafkaSource-style FLIP-27 source object without adaptation in the new API; homegrown source classes not built on the SourceProvider abstraction.","solutions":["Wrap a FLIP-27 source with DataStreamV2SourceUtils as the message suggests: env.fromSource(DataStreamV2SourceUtils.fromSource(flipp27Source, ...))","If the source is legacy SourceFunction-based, first migrate it to a FLIP-27 Source implementation, then wrap","Implement/extend the source abstractions the v2 environment recognizes (SourceProvider) instead of ad-hoc classes"],"exampleFix":"// before\nenv.fromSource(myLegacySourceFunction); // UnsupportedOperationException\n\n// after\norg.apache.flink.datastream.Source<String, ?, ?> src =\n        DataStreamV2SourceUtils.fromSource(kafkaSource, watermarkStrategy, \"kafka\");\nenv.fromSource(src);","handlingStrategy":"type-guard","validationCode":"// v2 env accepts sources built on FLIP-27; check before calling fromSource\nif (!(source instanceof org.apache.flink.datastream.Source)) {\n    source = DataStreamV2SourceUtils.fromSource(flip27Source, watermarkStrategy, \"name\");\n}\nenv.fromSource(source);","typeGuard":"boolean isV2Source(Object s) {\n    return s instanceof org.apache.flink.datastream.Source;\n}","tryCatchPattern":"catch (UnsupportedOperationException e) { /* unsupported source shape — wrap the source with DataStreamV2SourceUtils or implement SourceProvider; not transient, do not retry */ }","preventionTips":["Never pass legacy SourceFunction-based connectors or ad-hoc source objects to the v2 environment","Wrap FLIP-27 sources with DataStreamV2SourceUtils before fromSource","When porting v1 jobs, migrate each source to FLIP-27 first"],"tags":["datastream-v2","source","api-migration","flip-27"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}