{"record":{"id":"675eabd85ec23d4a","repo":"apache/flink","slug":"could-not-find-any-format-factory-for-identifier-675eab","errorCode":null,"errorMessage":"Could not find any format factory for identifier '%s' in the classpath.","messagePattern":"Could not find any format factory for identifier '(.+?)' in the classpath\\.","errorType":"validation","errorClass":"ValidationException","httpStatus":null,"severity":"error","filePath":"flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java","lineNumber":116,"sourceCode":"    // These mutable fields\n    private List<Map<String, String>> remainingPartitions;\n    private List<ResolvedExpression> filters;\n    private Long limit;\n    private int[][] projectFields;\n    private List<String> metadataKeys;\n    private DataType producedDataType;\n\n    public FileSystemTableSource(\n            ObjectIdentifier tableIdentifier,\n            DataType physicalRowDataType,\n            List<String> partitionKeys,\n            ReadableConfig tableOptions,\n            @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,\n            @Nullable DecodingFormat<DeserializationSchema<RowData>> deserializationFormat) {\n        super(tableIdentifier, physicalRowDataType, partitionKeys, tableOptions);\n        if (Stream.of(bulkReaderFormat, deserializationFormat).allMatch(Objects::isNull)) {\n            String identifier = tableOptions.get(FactoryUtil.FORMAT);\n            throw new ValidationException(\n                    String.format(\n                            \"Could not find any format factory for identifier '%s' in the classpath.\",\n                            identifier));\n        }\n        this.bulkReaderFormat = bulkReaderFormat;\n        this.deserializationFormat = deserializationFormat;\n        this.producedDataType = physicalRowDataType;\n    }\n\n    @Override\n    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {\n        // When this table has no partition, just return an empty source.\n        if (!partitionKeys.isEmpty() && getOrFetchPartitions().isEmpty()) {\n            return InputFormatProvider.of(new CollectionInputFormat<>(new ArrayList<>(), null));\n        }\n\n        // Resolve metadata and make sure to filter out metadata not in the producedDataType\n        final List<String> metadataKeys =","sourceCodeStart":98,"sourceCodeEnd":134,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java#L98-L134","documentation":"FileSystemTableSource's constructor checks that at least one reader format (bulkReaderFormat or deserializationFormat) is non-null. If both are null — meaning no decoding format was discovered for reading — it throws a ValidationException. The format identifier is retrieved from table options for the error message. This is a table-creation-time validation performed in the constructor.","triggerScenarios":"A CREATE TABLE with connector='filesystem' specifies a 'format' option for which no DecodingFormat factory is found in the classpath. For example, specifying 'format'='xyz' where no factory with identifier 'xyz' exists, or the format JAR is not on the classpath.","commonSituations":"The format JAR (e.g. flink-sql-parquet, flink-sql-csv) is not on the classpath when the table is created. Typo in the format identifier. Using a format identifier that no plugin provides. Classloader isolation issue preventing factory discovery via ServiceLoader.","solutions":["Add the required format JAR to the classpath of the SQL client or job (e.g. flink-sql-parquet.jar for parquet).","Verify the 'format' option value matches a known format identifier exactly (parquet, csv, json, orc, avro).","Check ServiceLoader discovery: ensure META-INF/services entries are present in the format JAR.","If running in a cluster, ensure the format JAR is shipped to all TaskManagers."],"exampleFix":"-- before\nCREATE TABLE source_t (a INT) WITH (\n  'connector'='filesystem',\n  'path'='file:///in',\n  'format'='unknownformat'\n);\n\n-- after\nCREATE TABLE source_t (a INT) WITH (\n  'connector'='filesystem',\n  'path'='file:///in',\n  'format'='parquet'\n);\n-- ensure flink-sql-parquet.jar is on the classpath","handlingStrategy":"validation","validationCode":"// Verify format JAR availability before table creation\nString formatId = options.get(\"format\");\nif (formatId == null) {\n    throw new ValidationException(\"'format' option is required for filesystem source\");\n}\nboolean found = ServiceLoader.load(Factory.class)\n    .stream()\n    .anyMatch(p -> p.get().identifier().equals(formatId));\nif (!found) {\n    throw new ValidationException(\n        \"No format factory for '\" + formatId + \"'. Add the format JAR to the classpath.\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Ensure the format JAR (flink-sql-parquet, flink-sql-csv, etc.) is on the classpath of the SQL client and cluster.","Verify the format identifier is spelled correctly and is a supported format.","Check META-INF/services entries in the format JAR for ServiceLoader discovery.","Use the same format JAR versions across all nodes in the cluster."],"tags":["table-source","format-factory","validation","classloader","flink"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}