apache/flink · error · RuntimeException

Cannot execute operator {}

Error message

Cannot execute operator {}

What it means

Thrown by CollectionExecutor.execute() when an operator in the plan does not match any of the six recognized operator type branches (BulkIterationBase, DeltaIterationBase, SingleInputOperator, DualInputOperator, GenericDataSourceBase, GenericDataSinkBase). The collection executor is used for local, in-JVM execution (CollectionEnvironment); it only supports a fixed set of operator types. An unrecognized type indicates either a custom operator subclass or a newer operator type the executor has not been updated to handle.

Source

Thrown at flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java:164

        if (operator instanceof BulkIterationBase) {
            result = executeBulkIteration((BulkIterationBase<?>) operator, jobInfo);
        } else if (operator instanceof DeltaIterationBase) {
            result = executeDeltaIteration((DeltaIterationBase<?, ?>) operator, jobInfo);
        } else if (operator instanceof SingleInputOperator) {
            result =
                    executeUnaryOperator(
                            (SingleInputOperator<?, ?, ?>) operator, superStep, jobInfo);
        } else if (operator instanceof DualInputOperator) {
            result =
                    executeBinaryOperator(
                            (DualInputOperator<?, ?, ?, ?>) operator, superStep, jobInfo);
        } else if (operator instanceof GenericDataSourceBase) {
            result = executeDataSource((GenericDataSourceBase<?, ?>) operator, superStep, jobInfo);
        } else if (operator instanceof GenericDataSinkBase) {
            executeDataSink((GenericDataSinkBase<?>) operator, superStep, jobInfo);
            result = Collections.emptyList();
        } else {
            throw new RuntimeException("Cannot execute operator " + operator.getClass().getName());
        }

        this.intermediateResults.put(operator, result);

        return result;
    }

    // --------------------------------------------------------------------------------------------
    //  Operator class specific execution methods
    // --------------------------------------------------------------------------------------------

    private <IN> void executeDataSink(GenericDataSinkBase<?> sink, int superStep, JobInfo jobInfo)
            throws Exception {
        Operator<?> inputOp = sink.getInput();
        if (inputOp == null) {
            throw new InvalidProgramException("The data sink " + sink.getName() + " has no input.");
        }

View on GitHub (pinned to 2f3c205e92)

Solutions

1. Switch from CollectionEnvironment to LocalEnvironment (createLocalEnvironment with actual execution) or a mini-cluster, which supports the full operator set.
2. If extending Flink internals, add a branch in CollectionExecutor.execute() for the new operator type.
3. Inspect the operator class name in the error message to identify which operator triggered it and whether it should be replaced with a standard equivalent.

Example fix

// before — using CollectionEnvironment which has limited operator support
env = new CollectionEnvironment();
// after — use LocalEnvironment which delegates to the full runtime
env = ExecutionEnvironment.createLocalEnvironment();

// Prefer LocalEnvironment for full operator support
ExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();
// Avoid CollectionEnvironment for jobs with complex/custom operators

When it happens

Trigger: Running a plan via ExecutionEnvironment.createLocalEnvironment() (or CollectionEnvironment) that contains an operator subclass not covered by the type checks. For example, a custom Operator subclass introduced in a plugin or extension. The DynamicPathCollector (error 375) has a similar but separate check.

Common situations: A developer tests a job locally with CollectionEnvironment but the job uses an operator only supported in the distributed runtime. Or a new operator type was added to the codebase but the CollectionExecutor was not updated. This is a programming/framework error, not a configuration issue.

Related errors

• Cannot handle operator type {}
• The data sink {} has no input.
• The unary operation {} has no input.
• The binary operation {} has no first input.
• The binary operation {} has no second input.

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