apache/flink · error · UnsupportedOperationException

Unsupported type of source, you could use DataStreamV2Source

Error message

Unsupported type of source, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source.

What it means

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.

Source

Thrown at flink-datastream/src/main/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImpl.java:195

                            WatermarkStrategy.noWatermarks(),
                            resolvedTypeInfo,
                            getParallelism(),
                            false);
            return StreamUtils.wrapWithConfigureHandle(
                    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 source, you could use DataStreamV2SourceUtils to wrap a FLIP-27 based source.");
        }
    }

    public Configuration getConfiguration() {
        return this.configuration;
    }

    public ExecutionConfig getExecutionConfig() {
        return executionConfig;
    }

    public int getParallelism() {
        return executionConfig.getParallelism();
    }

    public List<Transformation<?>> getTransformations() {
        return transformations;

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Wrap a FLIP-27 source with DataStreamV2SourceUtils as the message suggests: env.fromSource(DataStreamV2SourceUtils.fromSource(flipp27Source, ...))
2. If the source is legacy SourceFunction-based, first migrate it to a FLIP-27 Source implementation, then wrap
3. Implement/extend the source abstractions the v2 environment recognizes (SourceProvider) instead of ad-hoc classes

Example fix

// before
env.fromSource(myLegacySourceFunction); // UnsupportedOperationException

// after
org.apache.flink.datastream.Source<String, ?, ?> src =
        DataStreamV2SourceUtils.fromSource(kafkaSource, watermarkStrategy, "kafka");
env.fromSource(src);

// v2 env accepts sources built on FLIP-27; check before calling fromSource
if (!(source instanceof org.apache.flink.datastream.Source)) {
    source = DataStreamV2SourceUtils.fromSource(flip27Source, watermarkStrategy, "name");
}
env.fromSource(source);

boolean isV2Source(Object s) {
    return s instanceof org.apache.flink.datastream.Source;
}

catch (UnsupportedOperationException e) { /* unsupported source shape — wrap the source with DataStreamV2SourceUtils or implement SourceProvider; not transient, do not retry */ }

When it happens

Trigger: 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(...).

Common situations: 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.

Related errors

• This source reader does not support pausing or resuming spli
• Can not register process function transformation translator.
• Collection must not be empty
• Collection must not contain null elements
• Could not create TypeInformation for type {type}; please spe

AI-assisted analysis of apache/flink@2f3c205e92 (2026-08-14). Data as JSON: /api/errors/ed4a2048a6904a5a. Report an issue: GitHub.