{"record":{"id":"d43a037e5fae62d1","repo":"apache/flink","slug":"dynamicfilesplitenumerator-only-supports-batch-exe","errorCode":null,"errorMessage":"DynamicFileSplitEnumerator only supports batch execution.","messagePattern":"DynamicFileSplitEnumerator only supports batch execution\\.","errorType":"exception","errorClass":"UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/DynamicFileSplitEnumerator.java","lineNumber":198,"sourceCode":"        splitAssigner = splitAssignerFactory.create(splits);\n    }\n\n    @Override\n    public void addSplitsBack(List<SplitT> splits, int subtaskId) {\n        LOG.debug(\"Dynamic File Source Enumerator adds splits back: {}\", splits);\n        if (splitAssigner != null) {\n            List<FileSourceSplit> fileSplits = new ArrayList<>(splits);\n            // Only add back splits enumerating. A split may be filtered after it is assigned.\n            fileSplits.removeIf(s -> !allEnumeratingSplits.contains(s.splitId()));\n            // Added splits should be removed from assignedSplits for re-assignment\n            fileSplits.forEach(s -> assignedSplits.remove(s.splitId()));\n            splitAssigner.addSplits(fileSplits);\n        }\n    }\n\n    @Override\n    public PendingSplitsCheckpoint<SplitT> snapshotState(long checkpointId) {\n        throw new UnsupportedOperationException(\n                \"DynamicFileSplitEnumerator only supports batch execution.\");\n    }\n\n    @Override\n    public void handleSourceEvent(int subtaskId, int attemptNumber, SourceEvent sourceEvent) {\n        // Only recognize events that don't care attemptNumber\n        handleSourceEvent(subtaskId, sourceEvent);\n    }\n}\n","sourceCodeStart":180,"sourceCodeEnd":208,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/impl/DynamicFileSplitEnumerator.java#L180-L208","documentation":"The DynamicFileSplitEnumerator is a batch-only SplitEnumerator that supports dynamic filtering (DynamicFilteringEvent). Its snapshotState method unconditionally throws UnsupportedOperationException because checkpointing state — which is inherent to streaming execution — is not implemented for this enumerator. The enumerator is designed to enumerate and assign splits once in bounded (batch) mode, filtering splits based on dynamic filtering data received from the planner.","triggerScenarios":"The framework calls snapshotState(checkpointId) during a checkpoint barrier, which only occurs in streaming (continuous) execution mode. This happens when the FileSource is configured with dynamic filtering AND the job runs in STREAMING mode (e.g. when source.monitor-interval is set or the job is unbounded). The DynamicFileSplitEnumerator is instantiated internally by the table planner when dynamic filtering is enabled on a batch file source; if that job is accidentally or intentionally run as streaming, checkpointing triggers this method.","commonSituations":"Running a batch SQL/Table query with dynamic filtering (e.g. a dimension join that pushes down a filter) but the execution environment defaults to or is explicitly set to streaming mode. Migrating a batch pipeline to streaming without removing dynamic filtering. Enabling checkpointing on a job that uses dynamic file filtering.","solutions":["Ensure the job executes in BATCH mode: set execution.runtime-mode = BATCH in configuration, or call StreamExecutionEnvironment.setRuntimeMode(RuntimeExecutionMode.BATCH).","Remove dynamic filtering from the query if streaming execution is required, or restructure the query so the file source does not use DynamicFileSplitEnumerator.","Do not set source.monitor-interval on the file source when using dynamic filtering — its absence marks the source as bounded (batch)."],"exampleFix":"// before\nStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();\nenv.setRuntimeMode(RuntimeExecutionMode.STREAMING);\n// table uses dynamic filtering -> triggers snapshotState\n\n// after\nStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();\nenv.setRuntimeMode(RuntimeExecutionMode.BATCH);","handlingStrategy":"validation","validationCode":"// Before using dynamic filtering on a file source, verify batch mode\nif (env.getRuntimeMode() != RuntimeExecutionMode.BATCH) {\n    throw new IllegalStateException(\n        \"Dynamic filtering on file sources requires BATCH execution mode\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set execution.runtime-mode=BATCH when using dynamic filtering with file sources.","Do not set source.monitor-interval when dynamic filtering is active — it forces continuous (streaming) mode.","Test dynamic filtering pipelines in batch mode in CI to catch mode misconfigurations early."],"tags":["batch-mode","dynamic-filtering","checkpointing","file-source","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}