{"record":{"id":"13b02ccd39c3f95a","repo":"apache/flink","slug":"cannot-handle-operator-type","errorCode":null,"errorMessage":"Cannot handle operator type {}","messagePattern":"Cannot handle operator type (.+?)","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java","lineNumber":583,"sourceCode":"                    dynamicPathOperations.add(op);\n                } else if (dynamicPathOperations.contains(siop.getSecondInput())) {\n                    dynamicPathOperations.add(op);\n                } else {\n                    for (Operator<?> o : siop.getBroadcastInputs().values()) {\n                        if (dynamicPathOperations.contains(o)) {\n                            dynamicPathOperations.add(op);\n                            break;\n                        }\n                    }\n                }\n            } else if (op.getClass() == PartialSolutionPlaceHolder.class\n                    || op.getClass() == WorksetPlaceHolder.class\n                    || op.getClass() == SolutionSetPlaceHolder.class) {\n                dynamicPathOperations.add(op);\n            } else if (op instanceof GenericDataSourceBase) {\n                // skip\n            } else {\n                throw new RuntimeException(\n                        \"Cannot handle operator type \" + op.getClass().getName());\n            }\n        }\n    }\n\n    private class IterationRuntimeUDFContext extends RuntimeUDFContext\n            implements IterationRuntimeContext {\n\n        public IterationRuntimeUDFContext(\n                JobInfo jobInfo,\n                TaskInfo taskInfo,\n                ClassLoader classloader,\n                ExecutionConfig executionConfig,\n                Map<String, Future<Path>> cpTasks,\n                Map<String, Accumulator<?, ?>> accumulators,\n                OperatorMetricGroup metrics) {\n            super(jobInfo, taskInfo, classloader, executionConfig, cpTasks, accumulators, metrics);\n        }","sourceCodeStart":565,"sourceCodeEnd":601,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-core/src/main/java/org/apache/flink/api/common/operators/CollectionExecutor.java#L565-L601","documentation":"Thrown by the inner DynamicPathCollector class in CollectionExecutor during iteration execution when it encounters an operator that is not a SingleInputOperator, DualInputOperator, a known placeholder (PartialSolutionPlaceHolder, WorksetPlaceHolder, SolutionSetPlaceHolder), or a GenericDataSourceBase. This is a defensive check that traverses the iteration body to classify operators as on the dynamic path or not; an unrecognized type indicates an unsupported operator inside an iteration when using collection execution.","triggerScenarios":"Running an iteration (bulk or delta) via CollectionEnvironment where the iteration body contains an operator type the DynamicPathCollector does not recognize (e.g., GenericDataSinkBase inside an iteration, or a custom operator subclass). This is separate from the main execute() dispatch (error 364) — this check runs during the sub-plan traversal within iterations.","commonSituations":"A developer uses CollectionEnvironment to test an iterative algorithm, but the iteration body references an operator not supported by the collection executor's path analysis. The class name in the error message identifies the offending operator type.","solutions":["Switch from CollectionEnvironment to LocalEnvironment or a mini-cluster for iterative jobs with complex operator bodies.","Simplify the iteration body to use only standard operators (map, filter, join, etc.) that the DynamicPathCollector recognizes.","If extending Flink internals, update the DynamicPathCollector to handle the new operator type."],"exampleFix":"// before — complex operator in iteration body under CollectionEnvironment\nenv = new CollectionEnvironment();\nIterativeDataSet<Long> iter = source.iterate(10);\nDataSet<Long> step = iter.map(new RichMapFunction<>());  // may trigger unsupported type\n// after — use LocalEnvironment which supports the full operator set in iterations\nenv = ExecutionEnvironment.createLocalEnvironment();","handlingStrategy":"validation","validationCode":"// Use LocalEnvironment for iterative jobs with complex operator bodies\nExecutionEnvironment env = ExecutionEnvironment.createLocalEnvironment();\n// Avoid CollectionEnvironment for iterations with non-standard operators","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Prefer LocalEnvironment over CollectionEnvironment for iterative jobs.","Keep iteration bodies simple: use standard map/filter/join operators.","If extending Flink internals, update DynamicPathCollector to handle new operator types."],"tags":["collection-executor","iteration","operator-type","dynamic-path","flink-core"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}