{"record":{"id":"9b287aeab1bc57c4","repo":"apache/flink","slug":"cannot-execute-operator","errorCode":null,"errorMessage":"Cannot execute operator {}","messagePattern":"Cannot execute operator (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java","lineNumber":164,"sourceCode":"        if (operator instanceof BulkIterationBase) {\n            result = executeBulkIteration((BulkIterationBase<?>) operator, jobInfo);\n        } else if (operator instanceof DeltaIterationBase) {\n            result = executeDeltaIteration((DeltaIterationBase<?, ?>) operator, jobInfo);\n        } else if (operator instanceof SingleInputOperator) {\n            result =\n                    executeUnaryOperator(\n                            (SingleInputOperator<?, ?, ?>) operator, superStep, jobInfo);\n        } else if (operator instanceof DualInputOperator) {\n            result =\n                    executeBinaryOperator(\n                            (DualInputOperator<?, ?, ?, ?>) operator, superStep, jobInfo);\n        } else if (operator instanceof GenericDataSourceBase) {\n            result = executeDataSource((GenericDataSourceBase<?, ?>) operator, superStep, jobInfo);\n        } else if (operator instanceof GenericDataSinkBase) {\n            executeDataSink((GenericDataSinkBase<?>) operator, superStep, jobInfo);\n            result = Collections.emptyList();\n        } else {\n            throw new RuntimeException(\"Cannot execute operator \" + operator.getClass().getName());\n        }\n\n        this.intermediateResults.put(operator, result);\n\n        return result;\n    }\n\n    // --------------------------------------------------------------------------------------------\n    //  Operator class specific execution methods\n    // --------------------------------------------------------------------------------------------\n\n    private <IN> void executeDataSink(GenericDataSinkBase<?> sink, int superStep, JobInfo jobInfo)\n            throws Exception {\n        Operator<?> inputOp = sink.getInput();\n        if (inputOp == null) {\n            throw new InvalidProgramException(\"The data sink \" + sink.getName() + \" has no input.\");\n        }\n","sourceCodeStart":146,"sourceCodeEnd":182,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java#L146-L182","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Switch from CollectionEnvironment to LocalEnvironment (createLocalEnvironment with actual execution) or a mini-cluster, which supports the full operator set.","If extending Flink internals, add a branch in CollectionExecutor.execute() for the new operator type.","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."],"exampleFix":"// before — using CollectionEnvironment which has limited operator support\nenv = new CollectionEnvironment();\n// after — use LocalEnvironment which delegates to the full runtime\nenv = ExecutionEnvironment.createLocalEnvironment();","handlingStrategy":"validation","validationCode":"// Prefer LocalEnvironment for full operator support\nExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();\n// Avoid CollectionEnvironment for jobs with complex/custom operators","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Use LocalEnvironment instead of CollectionEnvironment for any non-trivial job.","Reserve CollectionEnvironment for simple, fully-supported operator chains.","If extending Flink internals with a new operator type, update CollectionExecutor.execute() dispatch."],"tags":["collection-executor","operator-type","local-execution","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}